[fix][broker] Fix topic compaction is failed after compactedLedger's all quorum is being recover

[TakaHiR07]

Fixes #21551

Motivation

Fix topic compaction is failed after compactedLedger's all quorum is being recover, which is described in issue.

Modifications

When CompactedTopicImpl try to open compactedLedger, use asyncOpenLedgerNoRecovery instead of asyncOpenLedger. Then compactedLedger can watch the zk node change and update the ledger metadata in broker.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: TakaHiR07#19

[codelipenghui]

@TakaHiR07 But the new solution will introduce a data inconsistent issue?

[zymap]

As you described in the issue, it seems the ledger doesn't replicate successfully to the new ensemble. The metadata is still the old one:

Read of ledger entry failed: L1705537 E0-E0, Sent to [bookie-1, bookie-2, bookie-3], Heard from [] : bitset = {}, Error = 'Bookie handle is not available'. First unread entry is (-1, rc = null)

It is still trying to read the old ensembles.

asyncOpenLedgerNoRecovery is not recommended because it could receive different responses on the different bookies.
For example, when you write 3 bookies and you need 2 bookie ack but only receive one ack. Using this API can read that entry from one bookie but can not read it from the another the failed write bookie. The write client know it's a failed write. But the read client can read it.

[TakaHiR07]

As you described in the issue, it seems the ledger doesn't replicate successfully to the new ensemble. The metadata is still the old one:

Read of ledger entry failed: L1705537 E0-E0, Sent to [bookie-1, bookie-2, bookie-3], Heard from [] : bitset = {}, Error = 'Bookie handle is not available'. First unread entry is (-1, rc = null)

It is still trying to read the old ensembles.

asyncOpenLedgerNoRecovery is not recommended because it could receive different responses on the different bookies. It violates the bookie's write consistent policy. For example, when you write 3 bookies and you need 2 bookie ack but only receive one ack. Using this API can read that entry from one bookie but can not read it from the another the failed write bookie. The write client know it's a failed write. But the read client can read it.

Replicate of auto-recovery is success, metadata on zk is the new one, but metadata on CompactTopicContext#ReadLedgerHandle is the old one. And I can read ledger directly from bookie-shell.

I guess compactedLedger is closed ledger, it should not be write and read at the same time. Therefore the read client should not be able to read the failed write entry ??

And I also have a question here. Why asyncOpenLedger not register the zk listener to update metadata change?

[zymap]

openLedger is a frequent operation, If every openLedger operation registers the listener, that would be a huge number.
If one broker can not communicate with zookeeper and doesn't close the compacted ledger, then a new broker opens the ledger. That would take some unexpected things.

[yebai1105]

I have the same problem, is there any new progress?

[TakaHiR07]

This problem has not been fixed and I hope to continue the review of this pr.

Firstly, I think the data inconsistent issue of asyncOpenLedgerNoRecovery is not exist.

• Because we can read the compactedLedger until ledger become Closed. There is no write client and read client exist at the same time. Just as the following code show, we close the ledger success and then do reader.acknowledgeCumulativeAsync, then COMPACTED_TOPIC_LEDGER_PROPERTY would be changed to the specific ledgerId, and then we asyncOpenLedgerNoRecovery the compactedLedger and the compactedTopicContext would be created.
• If the write client not finish, we can not create compactedTopicContext and can not create the read client.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/AbstractTwoPhaseCompactor.java

Lines 184 to 212 in 8a40b30

	private CompletableFuture<Long> phaseTwoSeekThenLoop(RawReader reader, MessageId from,
	MessageId to,
	MessageId lastReadId, Map<String, MessageId> latestForKey, BookKeeper bk,
	LedgerHandle ledger) {
	CompletableFuture<Long> promise = new CompletableFuture<>();
	reader.seekAsync(from).thenCompose((v) -> {
	Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
	CompletableFuture<Void> loopPromise = new CompletableFuture<>();
	phaseTwoLoop(reader, to, latestForKey, ledger, outstanding, loopPromise, MessageId.earliest);
	return loopPromise;
	}).thenCompose((v) -> closeLedger(ledger))
	.thenCompose((v) -> reader.acknowledgeCumulativeAsync(lastReadId,
	Map.of(COMPACTED_TOPIC_LEDGER_PROPERTY, ledger.getId())))
	.whenComplete((res, exception) -> {
	if (exception != null) {
	deleteLedger(bk, ledger).whenComplete((res2, exception2) -> {
	if (exception2 != null) {
	log.warn("Cleanup of ledger {} for failed", ledger, exception2);
	}
	// complete with original exception
	promise.completeExceptionally(exception);
	});
	} else {
	promise.complete(ledger.getId());
	}
	});
	return promise;
	}

Secondly, we only register listener for compactedLedger, the listener number is not huge.

@codelipenghui @zymap @codelipenghui @yebai1105 @coderzc

[lhotari]

possible related: apache/bookkeeper#4613