Issue 4200: fix flaky test DeferredSyncTest.testForceWillAdvanceLacOnlyUpToLastAcknoledgedWrite (#4234)

* Issue 4200: fix flaky test DeferredSyncTest.testForceWillAdvanceLacOnlyUpToLastAcknoledgedWrite

* add a comment

(cherry picked from commit 37708ad)

Author: tmzk1005

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

@@ -301,8 +301,15 @@ protected void suspendBookieForceLedgerAcks(BookieId address) {
     }
 
     protected void resumeBookieWriteAcks(BookieId address) {
-        suspendedBookiesForForceLedgerAcks.remove(address);
-        List<Runnable> pendingResponses = deferredBookieForceLedgerResponses.remove(address);
+        List<Runnable> pendingResponses;
+
+        // why use the BookieId instance as the object monitor? there is a date race problem if not
+        // see https://github.com/apache/bookkeeper/issues/4200
+        synchronized (address) {
+            suspendedBookiesForForceLedgerAcks.remove(address);
+            pendingResponses = deferredBookieForceLedgerResponses.remove(address);
+        }
+
         if (pendingResponses != null) {
             pendingResponses.forEach(Runnable::run);
         }
@@ -654,11 +661,17 @@ protected void setupBookieClientForceLedger() {
                     callback.forceLedgerComplete(BKException.Code.OK, ledgerId, bookieSocketAddress, ctx);
                 });
             };
-            if (suspendedBookiesForForceLedgerAcks.contains(bookieSocketAddress)) {
-                List<Runnable> queue = deferredBookieForceLedgerResponses.computeIfAbsent(bookieSocketAddress,
-                        (k) -> new CopyOnWriteArrayList<>());
-                queue.add(activity);
-            } else {
+            List<Runnable> queue = null;
+
+            synchronized (bookieSocketAddress) {
+                if (suspendedBookiesForForceLedgerAcks.contains(bookieSocketAddress)) {
+                    queue = deferredBookieForceLedgerResponses.computeIfAbsent(bookieSocketAddress,
+                            (k) -> new CopyOnWriteArrayList<>());
+                    queue.add(activity);
+                }
+            }
+
+            if (queue == null) {
                 activity.run();
             }
             return null;