Change rollCurrentLedgerIfFull logic to follow lazy creation of ledger

[mattisonchao]

Motivation

The original ledger creation design was lazy, meaning that the ledger was created when a new write operation was requested.

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

Lines 778 to 786 in ad2cc2d

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

However, the current rollCurrentLedgerIfFull logic is to create a ledger regardless of the condition (even if ManagedLedgerImpl#pendingAddEntries is empty)

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

Lines 1669 to 1692 in ad2cc2d

	@VisibleForTesting
	@Override
	public void rollCurrentLedgerIfFull() {
	log.info("[{}] Start checking if current ledger is full", name);
	if (currentLedgerEntries > 0 && currentLedgerIsFull()
	&& STATE_UPDATER.compareAndSet(this, State.LedgerOpened, 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());
	}
	}

Therefore, we need to change to lazy creation according to the original design.

Modifications

• Remove createLedgerAfterClosed invoke in the rollCurrentLedgerIfFull method.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• no-need-doc

[mattisonchao]

@eolivelli
Since this PR changes the behaviour of rollCurrentLedgerIfFull, the test ManagedCursorTest#testFindNewestMatchingAfterLedgerRollover needs some changes and I need your review.

[mattisonchao]

@codelipenghui @BewareMyPower @eolivelli @lhotari @michaeljmarshall @Technoboy-

This change affects some tests, PTAL :)