[Fix][broker] Fix NPE when ledger id not found in OpReadEntry

[mattisonchao]

Motivation

In current implementation, We have more than one potential NPE relate ManagedLedgerImpl#startReadOperationOnLedger method.

1. Unbox NPE

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

Lines 2230 to 2244 in b13d15c

	PositionImpl startReadOperationOnLedger(PositionImpl position, OpReadEntry opReadEntry) {
	Long ledgerId = ledgers.ceilingKey(position.getLedgerId());
	if (null == ledgerId) {
	opReadEntry.readEntriesFailed(new ManagedLedgerException.NoMoreEntriesToReadException("The ceilingKey(K key"
	+ ") method is used to return the least key greater than or equal to the given key, "
	+ "or null if there is no such key"), null);
	}
	if (ledgerId != position.getLedgerId()) {
	// The ledger pointed by this position does not exist anymore. It was deleted because it was empty. We need
	// to skip on the next available ledger
	position = new PositionImpl(ledgerId, 0);
	}
	return position;

According to this code, we can know that when the ledger id is null, we will fail OpReadEntry. However, there is no direct return here after failure. NPE will be thrown on line 2238. Because we want to unbox null.

2. null OpReadEntry context

According to the code above, we can see when we fail OpReadEntry, we pass a null value as context(line 2235). there is an NPE when the dispatcher gets the callback.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java

Lines 478 to 484 in 0975cdc

	private synchronized void internalReadEntriesFailed(ManagedLedgerException exception, Object ctx) {
	havePendingRead = false;
	ReadEntriesCtx readEntriesCtx = (ReadEntriesCtx) ctx;
	Consumer c = readEntriesCtx.getConsumer();
	readEntriesCtx.recycle();
	long waitTimeMillis = readFailureBackoff.next();

3. OpReadEntry#create

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

Lines 48 to 63 in 9376128

	public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPositionRef, int count,
	ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) {
	OpReadEntry op = RECYCLER.get();
	op.readPosition = cursor.ledger.startReadOperationOnLedger(readPositionRef, op);
	op.cursor = cursor;
	op.count = count;
	op.callback = callback;
	op.entries = Lists.newArrayList();
	if (maxPosition == null) {
	maxPosition = PositionImpl.LATEST;
	}
	op.maxPosition = maxPosition;
	op.ctx = ctx;
	op.nextReadPosition = PositionImpl.get(op.readPosition);
	return op;
	}

When we create OpReadEntry and then call ManagedCursor#startReadOperationOnLedger , assuming the ledger id is equal to null. At this point, we will fail this OpReadEntry. But at the current time, other parameters are not initialized. When we call OpReadEntry#readEntriesFailed. The cursor will be null and the NPE will be thrown.

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

Lines 90 to 94 in 9376128

	@Override
	public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
	cursor.readOperationCompleted();
	if (!entries.isEmpty()) {

Modifications

• Remove fail OpReadEntry logic in ManagedCursor#startReadOperationOnLedger, when ledger id equals null, we can return the original value. the ManagedLedger will validate if this operation is legal.

From the perspective of the overall design, the OpReadEntry is just a middle state object, that may have illegal value, we have to check this OpReadEntry is valid in ManagedLedger#asyncReadEntries(Current we already do that).

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc-not-needed
  (Please explain why)

[github-actions]

@mattisonchao:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[HQebupt]

/pulsarbot run-failure-checks

[mattisonchao]

After discussing with @Technoboy-, we may find why the ledgers.ceilingKey get Null; please consider this condition:

Precondition

• We have a ledger 12222:-1 at the ManagedLedger ledgers.
• We have an inactive cursor that read position is 12222:0.

Cases

1. When a new producer wants to produce a message on the topic but adds an entry to BK failed. It will prepare to close the current ledger and create a new one. because the current ledger entries are 0. So, we decided to delete it immediately. The logic shows below:

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

Lines 1618 to 1659 in 5455f4d

	synchronized void ledgerClosed(final LedgerHandle lh) {
	final State state = STATE_UPDATER.get(this);
	LedgerHandle currentLedger = this.currentLedger;
	if (currentLedger == lh && (state == State.ClosingLedger || state == State.LedgerOpened)) {
	STATE_UPDATER.set(this, State.ClosedLedger);
	} else if (state == State.Closed) {
	// The managed ledger was closed during the write operation
	clearPendingAddEntries(new ManagedLedgerAlreadyClosedException("Managed ledger was already closed"));
	return;
	} else {
	// In case we get multiple write errors for different outstanding write request, we should close the ledger
	// just once
	return;
	}
	long entriesInLedger = lh.getLastAddConfirmed() + 1;
	if (log.isDebugEnabled()) {
	log.debug("[{}] Ledger has been closed id={} entries={}", name, lh.getId(), entriesInLedger);
	}
	if (entriesInLedger > 0) {
	LedgerInfo info = LedgerInfo.newBuilder().setLedgerId(lh.getId()).setEntries(entriesInLedger)
	.setSize(lh.getLength()).setTimestamp(clock.millis()).build();
	ledgers.put(lh.getId(), info);
	} else {
	// The last ledger was empty, so we can discard it
	ledgers.remove(lh.getId());
	mbean.startDataLedgerDeleteOp();
	bookKeeper.asyncDeleteLedger(lh.getId(), (rc, ctx) -> {
	mbean.endDataLedgerDeleteOp();
	log.info("[{}] Delete complete for empty ledger {}. rc={}", name, lh.getId(), rc);
	}, null);
	}
	trimConsumedLedgersInBackground();
	maybeOffloadInBackground(NULL_OFFLOAD_PROMISE);
	if (!pendingAddEntries.isEmpty()) {
	// Need to create a new ledger to write pending entries
	createLedgerAfterClosed();
	}
	}

After line:1643, we may have empty ledgers.

2. In the meantime, the new subscription wants to read messages and create the OpReadEntry. but the ledgers are empty then NPE occur at startReadOperationOnLedger method.

/cc @Technoboy- @codelipenghui

[mattisonchao]

And because the ManagedLedger has the state to check if we can do something. So, I think the null check at OpReadEntry is enough.

[gaoran10]

LGTM

[gaoran10]

Maybe we could add a unit test to reproduce this problem, for example reading a position greater than max ledger id exists in ManagedLedgerImpl#ledgers? Or the ledgers is an empty set.

[mattisonchao]

@gaoran10 Got it, I will do that.

[mattisonchao]

Maybe we could add a unit test to reproduce this problem, for example reading a position greater than max ledger id exists in ManagedLedgerImpl#ledgers? Or the ledgers is an empty set.

Done, PTAL.

[Jason918]

move release/2.7.5 to #16966