Fix unit test

apache · Dec 27, 2019 · ef23a4b · ef23a4b
1 parent f659e8c
commit ef23a4b
Showing 1 changed file with 31 additions and 26 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1292,40 +1292,45 @@ public synchronized void updateLedgersIdsComplete(Stat stat) {
         if (log.isDebugEnabled()) {
             log.debug("[{}] Resending {} pending messages", name, pendingAddEntries.size());
         }
-        // Process all the pending addEntry requests
+
+        // Avoid use same OpAddEntry between different ledger handle
         int pendingSize = pendingAddEntries.size();
-        OpAddEntry op;
+        OpAddEntry existsOp;
         do {
-            op = pendingAddEntries.poll();
-            if (op != null) {
+            existsOp = pendingAddEntries.poll();
+            if (existsOp != null) {
                 // If op is used by another ledger handle, we need to close it and create a new one
-                if (op.ledger != null) {
-                    op.close();
-                    op = OpAddEntry.create(op.ml, op.data, op.callback, op.ctx);
+                if (existsOp.ledger != null) {
+                    existsOp.close();
+                    existsOp = OpAddEntry.create(existsOp.ml, existsOp.data, existsOp.callback, existsOp.ctx);
                 }
-                op.setLedger(currentLedger);
-                pendingAddEntries.add(op);
-                ++currentLedgerEntries;
-                currentLedgerSize += op.data.readableBytes();
+                existsOp.setLedger(currentLedger);
+                pendingAddEntries.add(existsOp);
+            }
+        } while (existsOp != null && --pendingSize > 0);
 
-                if (log.isDebugEnabled()) {
-                    log.debug("[{}] Sending {}", name, op);
-                }
+        // Process all the pending addEntry requests
+        for (OpAddEntry op : pendingAddEntries) {
+            ++currentLedgerEntries;
+            currentLedgerSize += op.data.readableBytes();
 
-                if (currentLedgerIsFull()) {
-                    STATE_UPDATER.set(this, State.ClosingLedger);
-                    op.setCloseWhenDone(true);
-                    op.initiate();
-                    if (log.isDebugEnabled()) {
-                        log.debug("[{}] Stop writing into ledger {} queue={}", name, currentLedger.getId(),
-                                pendingAddEntries.size());
-                    }
-                    break;
-                } else {
-                    op.initiate();
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Sending {}", name, op);
+            }
+
+            if (currentLedgerIsFull()) {
+                STATE_UPDATER.set(this, State.ClosingLedger);
+                op.setCloseWhenDone(true);
+                op.initiate();
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Stop writing into ledger {} queue={}", name, currentLedger.getId(),
+                            pendingAddEntries.size());
                 }
+                break;
+            } else {
+                op.initiate();
             }
-        } while (op != null && --pendingSize > 0);
+        }
     }
 
     // //////////////////////////////////////////////////////////////////////