Reduce metadataLock contention in LedgerHandle (#4549)

(cherry picked from commit b10aec2)

Author: lifepuzzlefun

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

@@ -1849,19 +1849,24 @@ void notifyWriteFailed(int index, BookieId addr) {
     }
 
     void maybeHandleDelayedWriteBookieFailure() {
+        Map<Integer, BookieId> toReplace = null;
         synchronized (metadataLock) {
             if (delayedWriteFailedBookies.isEmpty()) {
                 return;
             }
-            Map<Integer, BookieId> toReplace = new HashMap<>(delayedWriteFailedBookies);
+            toReplace = new HashMap<>(delayedWriteFailedBookies);
             delayedWriteFailedBookies.clear();
+        }
 
-            // Original intent of this change is to do a best-effort ensemble change.
-            // But this is not possible until the local metadata is completely immutable.
-            // Until the feature "Make LedgerMetadata Immutable #610" Is complete we will use
-            // handleBookieFailure() to handle delayed writes as regular bookie failures.
-            handleBookieFailure(toReplace);
+        if (toReplace.isEmpty()) {
+            return;
         }
+
+        // Original intent of this change is to do a best-effort ensemble change.
+        // But this is not possible until the local metadata is completely immutable.
+        // Until the feature "Make LedgerMetadata Immutable #610" Is complete we will use
+        // handleBookieFailure() to handle delayed writes as regular bookie failures.
+        handleBookieFailure(toReplace);
     }
 
     void handleBookieFailure(final Map<Integer, BookieId> failedBookies) {
@@ -1980,12 +1985,12 @@ void ensembleChangeLoop(List<BookieId> origEnsemble, Map<Integer, BookieId> fail
 
                         List<BookieId> newEnsemble = null;
                         Set<Integer> replaced = null;
+
+                        Map<Integer, BookieId> toReplace = null;
                         synchronized (metadataLock) {
                             if (!delayedWriteFailedBookies.isEmpty()) {
-                                Map<Integer, BookieId> toReplace = new HashMap<>(delayedWriteFailedBookies);
+                                toReplace = new HashMap<>(delayedWriteFailedBookies);
                                 delayedWriteFailedBookies.clear();
-
-                                ensembleChangeLoop(origEnsemble, toReplace);
                             } else {
                                 newEnsemble = getCurrentEnsemble();
                                 replaced = EnsembleUtils.diffEnsemble(origEnsemble, newEnsemble);
@@ -1994,6 +1999,11 @@ void ensembleChangeLoop(List<BookieId> origEnsemble, Map<Integer, BookieId> fail
                                 changingEnsemble = false;
                             }
                         }
+
+                        if (toReplace != null && !toReplace.isEmpty()) {
+                            ensembleChangeLoop(origEnsemble, toReplace);
+                        }
+
                         if (newEnsemble != null) { // unsetSuccess outside of lock
                             unsetSuccessAndSendWriteRequest(newEnsemble, replaced);
                         }