fix_topic_unavailable_error

Author: graysonzeng

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

@@ -146,6 +146,8 @@ private enum HandleState {
 
     ScheduledFuture<?> timeoutFuture = null;
 
+    private AtomicBoolean sendingCallbacks = new AtomicBoolean(false);
+
     @VisibleForTesting
     final Map<Integer, BookieId> delayedWriteFailedBookies =
         new HashMap<Integer, BookieId>();
@@ -1811,12 +1813,17 @@ void sendAddSuccessCallbacks() {
         // entries that have had all their responses come back
         PendingAddOp pendingAddOp;
 
+        if (!sendingCallbacks.compareAndSet(false, true)) {
+            return;
+        }
+
         while ((pendingAddOp = pendingAddOps.peek()) != null
                && !changingEnsemble) {
             if (!pendingAddOp.completed) {
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("pending add not completed: {}", pendingAddOp);
                 }
+                sendingCallbacks.set(false);
                 return;
             }
             // Check if it is the next entry in the sequence.
@@ -1825,6 +1832,7 @@ void sendAddSuccessCallbacks() {
                     LOG.debug("Head of the queue entryId: {} is not the expected value: {}", pendingAddOp.entryId,
                                pendingAddsSequenceHead + 1);
                 }
+                sendingCallbacks.set(false);
                 return;
             }
 
@@ -1838,6 +1846,7 @@ void sendAddSuccessCallbacks() {
             pendingAddOp.submitCallback(BKException.Code.OK);
         }
 
+        sendingCallbacks.set(false);
     }
 
     @VisibleForTesting