[fix][broker] Fix cannot cleanup expired ledger by trim ledgers

[Nicklee007]

Fixes #15473

Motivation

The pr #10087 has fixed some case expired data cannot cleanup, but in some other case the bug reappeared .

case 1 : the slowestReaderPosition be reset by other cursor operator before we can trim the expired ledger.

The topic producer produced some data and stoped in a time, then no more message will be add. When the maximumRolloverTimeMs reached, check the current ledger is full , closed the old ledger and created new ledger, after ledger create complete maybeUpdateCursorBeforeTrimmingConsumedLedger() will update cursor slowestReaderPosition to (newLedgerId,-1) and trigger trimConsumedLedgersInBackground() immediately

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Line 2284 in a1fb200

lastAckedPosition = new PositionImpl(nextPointedLedger.getLedgerId(), -1);

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Lines 2295 to 2296 in a1fb200

	log.info("Reset cursor:{} to {} since ledger consumed completely", cursor, lastAckedPosition);
	updateCursor((ManagedCursorImpl) cursor, lastAckedPosition);

But almost the message retentionTimeMs is longer than maximumRolloverTimeMs, then the ledger can not be clean by trimConsumedLedgersInBackground at this time running.
And then if we have a flink task consumer the topic( use reader to consumer and mark position by another durable cursor), the client will always reset the durable cursor to the position (olderLedgerId, entryNum +1)
internalResetCursor invoke internalAsyncMarkDelete invoke ledger.updateCursor to reset slowestReaderPosition to (olderLedgerId, entryNum ), when reached the next retentionCheckIntervalInSeconds to invoke trimConsumedLedgersInBackground ,but the slowestReaderPosition has been reset to the (olderLedgerId, entryNum ), then we can not clean the old ledger in any reach 'retentionCheckIntervalInSeconds' time.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Line 2474 in ac6bd3c

for (LedgerInfo ls : ledgers.headMap(slowestReaderLedgerId, false).values()) {

Also the old ledger is not expired when after created ledger trigger trimConsumedLedgersInBackground ,another possibility cursor operator is the markDeleteLimiter cause the consumer last mark-delete operations is Dirty and need mark delete by flushCursorsTask also will invokeledger.updateCursor(ManagedCursorImpl.this, mdEntry.newPosition) to (olderLedgerId, entryNum ) . After this time we also cloud not clean the old ledger by trimConsumedLedgersInBackground.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

Lines 3132 to 3138 in ac6bd3c

	void flush() {
	if (!isDirty) {
	return;
	}
	isDirty = false;
	asyncMarkDelete(lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, new MarkDeleteCallback() {

case2: have not a condition to set slowestReaderPosition to the correct position.

In this case, the most common is the producer stoped in one time and we have durable cursor, but we have not active consumer, the cursor position move forward by MessageExpiryMonitor,maybe the cursor mark delete position is before a few ledger , and the last time create ledger and invoke maybeUpdateCursorBeforeTrimmingConsumedLedger maybe not set slowestReaderPosition to the newest ledger (newLedgerId,-1) , after then MessageExpiryMonitor only can reset slowestReaderPosition to(olderLedgerId, entryNum ) . After this time we also cloud not clean the old ledger by trimConsumedLedgersInBackground.

Modifications

1. when invoke checkConsumedLedgers , we use maybeUpdateCursorBeforeTrimmingConsumedLedger() to check if ledger consumed completely and reset cursor to next ledger, then the ConsumedLedgersMonitor can trim the old ledger data.
2. PR #14672 change rollCurrentLedgerIfFull logic to follow lazy creation of ledger, we need create new ledger to reset slowestReaderPosition to the new empty ledger. So we need add invoke createLedgerAfterClosed() in rollCurrentLedgerIfFull. -- PR Change rollCurrentLedgerIfFull logic to follow lazy creation of ledger #14672 has revert.
3. add some unit test.

Documentation

Check the box below or label this PR directly.
Need to update docs?

• doc-not-needed
  (Please explain why)

[poorbarcode]

Good catch. This patch solved the problem to some extent, I think the solution is that the newly created persistent subscription 'cursor position' is always greater than 'message-TTL' and the 'message-TTL' policy is enabled.

I have two questions:

• 1

And then if we have a flink task consumer the topic( use reader to consumer and mark position by another durable cursor), the client will always reset the durable cursor to the position.

why not use nodurable Reader ?

• 2

ledgerClosed(lh);
createLedgerAfterClosed();

This should ensure that a new-leadger is always created when the old-leadger is closed. If this is necessary, a comment should be added to remind others later, better have a unit test to ensure that.

[Nicklee007]

Good catch. This patch solved the problem to some extent, I think the solution is that the newly created persistent subscription 'cursor position' is always greater than 'message-TTL' and the 'message-TTL' policy is enabled.

I have two questions:

• 1

And then if we have a flink task consumer the topic( use reader to consumer and mark position by another durable cursor), the client will always reset the durable cursor to the position.

why not use nodurable Reader ?

• 2

ledgerClosed(lh);
createLedgerAfterClosed();

This should ensure that a new-leadger is always created when the old-leadger is closed. If this is necessary, a comment should be added to remind others later, better have a unit test to ensure that.

@poorbarcode yes, but only enable 'message-TTL' policy is not enough. As the case 2, if we have subscription cursor and not active consumer, the MessageExpiryMonitor only can reset slowestReaderPosition to(olderLedgerId, entryNum ), cannot move the cursor to the newest empty ledger, we move slowestReaderPosition through the ConsumedLedgersMonitor to check if the ledger consumed completely, we'll reset slowestReaderPosition and then trigger trimConsumedLedgersInBackground in one task , this will be more effective and probability of slowestReaderPosition value being covered is lower, even slowestReaderPosition is covered ,we could reset it to newest position in next MessageExpiryMonitor time.

why not use nodurable Reader ?
Because we have many flink job use the StreamNative's pulsar-connector which use a reader to consume data and need mark consumer position by another durable cursor when flink job restart which will be used. So there are two cursor.

[Nicklee007]

@mattisonchao
For clean expire ledger data, and the loaded topic need at least one ledger, this PR changes the behavior of rollCurrentLedgerIfFull, readd createLedgerAfterClosed() in rollCurrentLedgerIfFull I need your review.
I'll change the unit test later.

[Technoboy-]

Hi @Nicklee007
Could you write a test to reproduce this case?

[Nicklee007]

Hi @Nicklee007 Could you write a test to reproduce this case?

@Technoboy- added unit test for the case 2, PTAL.

[Nicklee007]

@mattisonchao For clean expire ledger data, and the loaded topic need at least one ledger, this PR changes the behavior of rollCurrentLedgerIfFull, readd createLedgerAfterClosed() in rollCurrentLedgerIfFull I need your review. I'll change the unit test later.

@mattisonchao hi, I have revert the unit test modify in the PR 14672, PTAL.

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[BewareMyPower]

I think we still need to revert #14672 first (#16806). Because this PR just checks whether any cursor has been updated in the the periodically BrokerService#checkConsumedLedgers call based on #16806, which restores the original behavior (same with branch-2.10 and earlier) that a new ledger will be created after closing the current ledger.

[Nicklee007]

I think we still need to revert #14672 first (#16806). Because this PR just checks whether any cursor has been updated in the the periodically BrokerService#checkConsumedLedgers call based on #16806, which restores the original behavior (same with branch-2.10 and earlier) that a new ledger will be created after closing the current ledger.

Ok, waiting #14672 reverted, I'll change this PR.

[mattisonchao]

Hi, @Nicklee007
The PR #14672 already reverted. please continue work. thanks a lot!

[Nicklee007]

@mattisonchao Changed this PR, PTAL.

[github-actions]

@Nicklee007 Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

[github-actions]

The pr had no activity for 30 days, mark with Stale label.