Fixed LedgerHandle#ensembleChangeLoop.

shustsud · shustsud · commit f47db3d343c4 · 2024-02-13T16:09:07.000+09:00
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -1994,11 +1994,14 @@ void ensembleChangeLoop(List<BookieId> origEnsemble, Map<Integer, BookieId> fail
                                 replaced = EnsembleUtils.diffEnsemble(origEnsemble, newEnsemble);
                                 LOG.info("New Ensemble: {} for ledger: {}", newEnsemble, ledgerId);
 
+                                // Since changingEnsemble is true, processing in #sendAddSuccessCallbacks() is skipped.
+                                unsetSuccessAndSendWriteRequest(newEnsemble, replaced);
                                 changingEnsemble = false;
                             }
                         }
-                        if (newEnsemble != null) { // unsetSuccess outside of lock
-                            unsetSuccessAndSendWriteRequest(newEnsemble, replaced);
+                        if (newEnsemble != null) {
+                            // After changingEnsemble is changed to false, call #sendAddSuccessCallbacks().
+                            sendAddSuccessCallbacks();
                         }
                     }
             }, clientCtx.getMainWorkerPool().chooseThread(ledgerId));

Original file line number	Diff line number	Diff line change
`@@ -1994,11 +1994,14 @@ void ensembleChangeLoop(List<BookieId> origEnsemble, Map<Integer, BookieId> fail`
`1994`	`1994`	`replaced = EnsembleUtils.diffEnsemble(origEnsemble, newEnsemble);`
`1995`	`1995`	`LOG.info("New Ensemble: {} for ledger: {}", newEnsemble, ledgerId);`
`1996`	`1996`
	`1997`	`+ // Since changingEnsemble is true, processing in #sendAddSuccessCallbacks() is skipped.`
	`1998`	`+ unsetSuccessAndSendWriteRequest(newEnsemble, replaced);`
`1997`	`1999`	`changingEnsemble = false;`
`1998`	`2000`	`}`
`1999`	`2001`	`}`
`2000`		`- if (newEnsemble != null) { // unsetSuccess outside of lock`
`2001`		`- unsetSuccessAndSendWriteRequest(newEnsemble, replaced);`
	`2002`	`+ if (newEnsemble != null) {`
	`2003`	`+ // After changingEnsemble is changed to false, call #sendAddSuccessCallbacks().`
	`2004`	`+ sendAddSuccessCallbacks();`
`2002`	`2005`	`}`
`2003`	`2006`	`}`
`2004`	`2007`	`}, clientCtx.getMainWorkerPool().chooseThread(ledgerId));`