Fix lost message issue due to ledger rollover.

[Technoboy-]

Motivation

pre-require : User config managedLedgerMaxLedgerRolloverTimeMinutes > 0.

Then, if ManagedLedger creates a ledger in the below case :

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

Lines 778 to 785 in 7998c44

	} else if (state == State.ClosedLedger) {
	// No ledger and no pending operations. Create a new ledger
	log.info("[{}] Creating a new ledger", name);
	if (STATE_UPDATER.compareAndSet(this, State.ClosedLedger, State.CreatingLedger)) {
	this.lastLedgerCreationInitiationTimestamp = System.currentTimeMillis();
	mbean.startDataLedgerCreateOp();
	asyncCreateLedger(bookKeeper, config, digestType, this, Collections.emptyMap());
	}

The ManagedLedger state now is CreatingLedger . At this moment, rollover is triggered, it will set the state to ClosingLedger (line-1674)

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

Lines 1671 to 1692 in 7998c44

	public void rollCurrentLedgerIfFull() {
	log.info("[{}] Start checking if current ledger is full", name);
	if (currentLedgerEntries > 0 && currentLedgerIsFull()) {
	STATE_UPDATER.set(this, State.ClosingLedger);
	currentLedger.asyncClose(new AsyncCallback.CloseCallback() {
	@Override
	public void closeComplete(int rc, LedgerHandle lh, Object o) {
	checkArgument(currentLedger.getId() == lh.getId(), "ledgerId %s doesn't match with "
	+ "acked ledgerId %s", currentLedger.getId(), lh.getId());
	if (rc == BKException.Code.OK) {
	log.debug("Successfully closed ledger {}", lh.getId());
	} else {
	log.warn("Error when closing ledger {}. Status={}", lh.getId(), BKException.getMessage(rc));
	}
	ledgerClosed(lh);
	createLedgerAfterClosed();
	}
	}, System.nanoTime());
	}
	}

And invoke createLedgerAfterClosed -> isNeededCreateNewLedgerAfterCloseLedger is passing,
because isNeededCreateNewLedgerAfterCloseLedger checks CreatingLedger and LedgerOpened state, but the current state is ClosingLedger:

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

Lines 1648 to 1667 in 7998c44

	synchronized void createLedgerAfterClosed() {
	if (isNeededCreateNewLedgerAfterCloseLedger()) {
	log.info("[{}] Creating a new ledger", name);
	STATE_UPDATER.set(this, State.CreatingLedger);
	this.lastLedgerCreationInitiationTimestamp = System.currentTimeMillis();
	mbean.startDataLedgerCreateOp();
	// Use the executor here is to avoid use the Zookeeper thread to create the ledger which will lead
	// to deadlock at the zookeeper client, details to see https://github.com/apache/pulsar/issues/13736
	this.executor.execute(() ->
	asyncCreateLedger(bookKeeper, config, digestType, this, Collections.emptyMap()));
	}
	}
	boolean isNeededCreateNewLedgerAfterCloseLedger() {
	final State state = STATE_UPDATER.get(this);
	if (state != State.CreatingLedger && state != State.LedgerOpened) {
	return true;
	}
	return false;
	}

So it will also create a new ledger. And result in the message being lost. It's the same result with #12221

Detail log:

2022-03-04T01:41:49.908333235+02:00 2022-03-03T23:41:49,907 [BookKeeperClientWorker-OrderedExecutor-3-0] INFO  org.apache.bookkeeper.mledger.impl.OpAddEntry - [public/default/persistent/task_topic-partition-0] Closing ledger 1846355 for being full
2022-03-04T01:41:49.921447018+02:00 2022-03-03T23:41:49,920 [BookKeeperClientWorker-OrderedExecutor-3-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Creating a new ledger
2022-03-04T01:41:49.921482783+02:00 2022-03-03T23:41:49,921 [BookKeeperClientWorker-OrderedExecutor-3-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/itomdipostload/persistent/task_topic-partition-0] Creating ledger, metadata: {component=[109, 97, 110, 97, 103, 101, 100, 45, 108, 101, 100, 103, 101, 114], pulsar/managed-ledger=[112, 117, 98, 108, 105, 99, 47, 105, 116, 111, 109, 100, 105, 112, 111, 115, 116, 108, 111, 97, 100, 47, 112, 101, 114, 115, 105, 115, 116, 101, 110, 116, 47, 100, 105, 95, 112, 111, 115, 116, 108, 111, 97, 100, 95, 116, 97, 115, 107, 95, 116, 111, 112, 105, 99, 45, 112, 97, 114, 116, 105, 116, 105, 111, 110, 45, 48], application=[112, 117, 108, 115, 97, 114]} - metadata ops timeout : 60 seconds
2022-03-04T01:41:49.926896931+02:00 2022-03-03T23:41:49,926 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Start checking if current ledger is full
2022-03-04T01:41:49.926944096+02:00 2022-03-03T23:41:49,926 [BookKeeperClientWorker-OrderedExecutor-3-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Creating a new ledger
2022-03-04T01:41:49.927043799+02:00 2022-03-03T23:41:49,926 [BookKeeperClientWorker-OrderedExecutor-3-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Creating ledger, metadata: {component=[109, 97, 110, 97, 103, 101, 100, 45, 108, 101, 100, 103, 101, 114], pulsar/managed-ledger=[112, 117, 98, 108, 105, 99, 47, 105, 116, 111, 109, 100, 105, 112, 111, 115, 116, 108, 111, 97, 100, 47, 112, 101, 114, 115, 105, 115, 116, 101, 110, 116, 47, 100, 105, 95, 112, 111, 115, 116, 108, 111, 97, 100, 95, 116, 97, 115, 107, 95, 116, 111, 112, 105, 99, 45, 112, 97, 114, 116, 105, 116, 105, 111, 110, 45, 48], application=[112, 117, 108, 115, 97, 114]} - metadata ops timeout : 60 seconds
2022-03-04T01:41:49.936780564+02:00 2022-03-03T23:41:49,936 [main-EventThread] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Created new ledger 1846376
2022-03-04T01:41:49.944880616+02:00 2022-03-03T23:41:49,944 [main-EventThread] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/task_topic-partition-0] Created new ledger 1846377

Modification

• Rollover only with ledger LedgerOpened state.
• Modify log to help troubleshoot problem.

Documentation

• no-need-doc

[eolivelli]

good catch and great work

[hangc0276]

Nice Catch.

[mattisonchao]

Great Catch!