issue 4109: Fix Flaky-test: HandleFailuresTest.testHandleFailureBookieNotInWriteSet

[AnonHxy]

Descriptions of the changes in this PR:

Master Issue: #4109

Motivation

Fix Flaky-test: HandleFailuresTest.testHandleFailureBookieNotInWriteSet

When we call b1Delay.completeExceptionally(new BKException.BKWriteException()) at line480(code1), the preWriteHook will complete with exception and then do some actions in the choosen thread(code2), e.g., put the failure bookie to delayedWriteFailedBookies(code3). So the delayedWriteFailedBookies update is async.

However, lh.appendAsync("entry2".getBytes()))(Line483 in Code4) is invoked in main thread. So when appendAsync execute, the delayedWriteFailedBookies could be not updated yet, and changeInProgress.complete(null)(code5) will never be invoked.

We can reproduce this flasy-test if we put Thread.sleep(5000) just before line289 at code6.

code1:

bookkeeper/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java

Line 480 in 907c273

b1Delay.completeExceptionally(new BKException.BKWriteException());

code2:

bookkeeper/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookieClient.java

Lines 179 to 186 in 907c273

	.whenCompleteAsync((res, ex) -> {
	if (ex != null) {
	cb.writeComplete(BKException.getExceptionCode(ex, BKException.Code.WriteException),
	ledgerId, entryId, addr, ctx);
	} else {
	cb.writeComplete(BKException.Code.OK, ledgerId, entryId, addr, ctx);
	}
	}, executor.chooseThread(ledgerId));

code3:

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java

Lines 1848 to 1852 in 907c273

	void notifyWriteFailed(int index, BookieId addr) {
	synchronized (metadataLock) {
	delayedWriteFailedBookies.put(index, addr);
	}
	}

code4:

bookkeeper/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java

Lines 480 to 483 in 907c273

	b1Delay.completeExceptionally(new BKException.BKWriteException());
	log.info("write second entry, should have enough bookies, but blocks completion on failure handling");
	CompletableFuture<?> e2 = lh.appendAsync("entry2".getBytes());

code5:

bookkeeper/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java

Lines 469 to 472 in 907c273

	clientCtx.getMockLedgerManager().setPreWriteHook((ledgerId, metadata) -> {
	changeInProgress.complete(null);
	return blockEnsembleChange;
	});

code6:

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java

Lines 282 to 290 in 907c273

	if (completed) {
	if (rc != BKException.Code.OK) {
	// Got an error after satisfying AQ. This means we are under replicated at the create itself.
	// Update the stat to reflect it.
	clientCtx.getClientStats().getAddOpUrCounter().inc();
	if (!clientCtx.getConf().disableEnsembleChangeFeature.isAvailable()
	&& !clientCtx.getConf().delayEnsembleChange) {
	lh.notifyWriteFailed(bookieIndex, addr);
	}

Changes

(Describe: what changes you have made)

[AnonHxy]

@hangc0276 @StevenLuMT @eolivelli @dlg99 PTAL

[hangc0276]

Nice catch!