[FIX] Fix race conditions in Auditor bookie change detection

[lhotari]

Motivation

While investigating a CI flakiness in BookieAutoRecoveryTest (PR #4711) — where tests waiting up to 60 s for ledgers to be marked as underreplicated would time out — several related race conditions in Auditor were found that could allow a bookie departure to go undetected until the next periodic audit run (default: 86400 s).

Analysis of the surefire logs for the failing CI run showed a ~70-second silence from the auditor after a bookie was killed, followed by a new auditor election that eventually caught the departure. This pointed to the auditor's bookie-change detection path rather than a test-setup issue.

Races fixed

Race 1 – watcher-provided bookie set was discarded

watchBookieChanges() registered bookies -> submitAuditTask() as the ZK listener, ignoring the current bookie set already delivered by the callback. Inside submitAuditTask() a fresh getAvailableBookies() ZK call was made from an executor lambda that could execute well after the watcher fired — for example while AuditorCheckAllLedgersTask was occupying the single-threaded executor. In that window the two snapshots of the cluster could diverge, causing bookiesToBeAudited to remain empty and the departure to be silently skipped.

Fix: capture pendingWritableBookies / pendingReadOnlyBookies atomically in each watcher callback and consume them in the extracted runAuditTask(). Fall back to a live getAvailableBookies() call only when one watcher type has not yet fired (to avoid an unnecessary ZK round-trip for the already-known half).

Race 2 – duplicate audit tasks could be queued per change type

Rapid successive watcher callbacks (or a slow executor) could enqueue many identical audit tasks, each making its own ZK calls and risking interleaved mutations to knownBookies / bookiesToBeAudited.

Fix: replace the original AtomicBoolean queued-flag with an AtomicInteger counter (writableAuditTaskCount, readOnlyAuditTaskCount). The counter tracks running + queued tasks per change type and is capped at MAX_AUDIT_TASKS_PER_TYPE = 2 (one in-progress + one queued). The queued task always reads the latest pendingWritableBookies / pendingReadOnlyBookies when it starts, so queuing more than one additional task would only duplicate work without improving correctness.

Race 3 – knownBookies initialised after watcher registration

start() called watchBookieChanges() before assigning knownBookies. A watcher callback firing immediately could race with the subsequent knownBookies = admin.getAllBookies() assignment.

Fix: swap the initialisation order so knownBookies is populated before watchBookieChanges() is called.

Race 4 – submitAuditTask() could not detect bookies that registered and died after auditor startup

submitAuditTask() (used by tests to explicitly trigger and wait for an audit) submitted runAuditTask(), which detects bookie losses by comparing pendingWritableBookies against knownBookies. knownBookies is seeded from admin.getAllBookies() at auditor startup, so any bookie that registered after the auditor started and died before being processed by runAuditTask() was never added to knownBookies and was invisible to the subtraction-based detection — regardless of the task cap.

This caused testEmptyLedgerLosesQuorumEventually to fail when run after testNoSuchLedgerExists: bookies 33105 and 36375 registered after auditor 42899 started; two watcher tasks were queued and were both blocked behind checkAllLedgers (112 ms); bookie 36375 died during that window; by the time the tasks ran, pendingWritableBookies correctly reflected {42899, 33105} but since 36375 was never in knownBookies the loss went undetected. Increasing MAX_AUDIT_TASKS_PER_TYPE does not help here — all tasks read pendingWritableBookies.get() when they execute, so they always see the latest snapshot (without 36375), and no amount of additional queued tasks changes that.

Fix: submitAuditTask() now submits auditorBookieCheckTask.startAudit(false) — the full ledger-metadata-based audit — rather than runAuditTask(). The full scan detects any bookie that appears in ledger ensembles but is no longer available, regardless of knownBookies state.

The runAuditTask() / knownBookies mechanism remains correct and sufficient for its intended purpose: detecting losses for bookies that were already known. The register-and-die edge case is an inherent limitation of the lightweight watcher-based approach and is handled by the periodic auditorBookieCheckTask and by explicit submitAuditTask() calls.

Changes summary

What	Change
Auditor.knownBookies	Add volatile; initialise before watchBookieChanges()
pendingWritableBookies, pendingReadOnlyBookies	New AtomicReference<Set<String>> fields holding the latest bookie set per watcher type
writableAuditTaskCount, readOnlyAuditTaskCount	New AtomicInteger counters (running+queued per type, capped at 2) replacing the original AtomicBoolean flags
watchBookieChanges()	Populate the pending sets and call submitAuditTaskForBookieChange(boolean)
submitAuditTaskForBookieChange(boolean)	Queues runAuditTask only when the per-type count allows
submitAuditTask()	Now submits auditorBookieCheckTask.startAudit(false) (full audit via ledger metadata) instead of runAuditTask()
runAuditTask()	Uses watcher-captured bookie sets; falls back to live ZK only when one watcher type is absent
start()	Initialise knownBookies before calling watchBookieChanges()

Testing

Existing tests in BookieAutoRecoveryTest exercise all affected paths and continue to pass. The previously flaky test (testEmptyLedgerLosesQuorumEventually) now passes reliably when run as part of the full test class.

[lhotari]

Although this fixes the flaky tests, there are multiple remaining race conditions in the auditor which could make it inoperational. For example, this doesn't look right for all cases:

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java

Lines 132 to 139 in 2eb70b1

	if (throwable != null) {
	if (firstRunFuture.isDone()) {
	scheduleWatchTask(ZK_CONNECT_BACKOFF_MS);
	} else {
	firstRunFuture.completeExceptionally(throwable);
	}
	return;
	}

This is also surprising:

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java

Lines 422 to 427 in 2eb70b1

	watchReadOnlyBookiesTask = new WatchTask(bookieReadonlyRegistrationPath, f);
	f = f.whenComplete((value, cause) -> {
	if (null != cause) {
	unwatchReadOnlyBookies(listener);
	}
	});

If the first time fails, registering the watcher would be silently ignored.