Avoid adding duplicated BrokerEntryMetadata (#12018)

BewareMyPower · web-flow · commit 9d44617a204f · 2021-09-18T20:43:02.000+08:00
### Motivation When the Pulsar cluster enables broker entry metadata, sometimes there're some corrupted entries. See streamnative/kop#442 for example. It's because the broker entry metadata has been added twice. This bug might be introduced from #9039 https://github.com/apache/pulsar/blob/9b7c3275c904ac1e6a8ef67487a10a0506bb2c58/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1516-L1518 It happened during a managed ledger's rollover while there're some pending `OpAddEntry`s in `updateLedgersIdsComplete`, only the ledger id should be updated and the data of `OpAddEntry` should not be modified. ### Modifications Only call `beforeAddEntry` for once at the beginning of `internalAsyncAddEntry`.
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
@@ -717,6 +717,9 @@ public void asyncAddEntry(ByteBuf buffer, int numberOfMessages, AddEntryCallback
     }
 
     private synchronized void internalAsyncAddEntry(OpAddEntry addOperation) {
+        if (!beforeAddEntry(addOperation)) {
+            return;
+        }
         pendingAddEntries.add(addOperation);
         final State state = STATE_UPDATER.get(this);
         if (state == State.Fenced) {
@@ -779,10 +782,7 @@ private synchronized void internalAsyncAddEntry(OpAddEntry addOperation) {
                 addOperation.setCloseWhenDone(true);
                 STATE_UPDATER.set(this, State.ClosingLedger);
             }
-            // interceptor entry before add to bookie
-            if (beforeAddEntry(addOperation)) {
-                addOperation.initiate();
-            }
+            addOperation.initiate();
         }
     }
 
@@ -1513,9 +1513,7 @@ public synchronized void updateLedgersIdsComplete(Stat stat) {
                     ReferenceCountUtil.release(existsOp.data);
                 }
                 existsOp.setLedger(currentLedger);
-                if (beforeAddEntry(existsOp)) {
-                    pendingAddEntries.add(existsOp);
-                }
+                pendingAddEntries.add(existsOp);
             }
         } while (existsOp != null && --pendingSize > 0);
 

Original file line number	Diff line number	Diff line change
`@@ -717,6 +717,9 @@ public void asyncAddEntry(ByteBuf buffer, int numberOfMessages, AddEntryCallback`
`717`	`717`	`}`
`718`	`718`
`719`	`719`	`private synchronized void internalAsyncAddEntry(OpAddEntry addOperation) {`
	`720`	`+ if (!beforeAddEntry(addOperation)) {`
	`721`	`+ return;`
	`722`	`+ }`
`720`	`723`	`pendingAddEntries.add(addOperation);`
`721`	`724`	`final State state = STATE_UPDATER.get(this);`
`722`	`725`	`if (state == State.Fenced) {`
`@@ -779,10 +782,7 @@ private synchronized void internalAsyncAddEntry(OpAddEntry addOperation) {`
`779`	`782`	`addOperation.setCloseWhenDone(true);`
`780`	`783`	`STATE_UPDATER.set(this, State.ClosingLedger);`
`781`	`784`	`}`
`782`		`- // interceptor entry before add to bookie`
`783`		`- if (beforeAddEntry(addOperation)) {`
`784`		`- addOperation.initiate();`
`785`		`- }`
	`785`	`+ addOperation.initiate();`
`786`	`786`	`}`
`787`	`787`	`}`
`788`	`788`
`@@ -1513,9 +1513,7 @@ public synchronized void updateLedgersIdsComplete(Stat stat) {`
`1513`	`1513`	`ReferenceCountUtil.release(existsOp.data);`
`1514`	`1514`	`}`
`1515`	`1515`	`existsOp.setLedger(currentLedger);`
`1516`		`- if (beforeAddEntry(existsOp)) {`
`1517`		`- pendingAddEntries.add(existsOp);`
`1518`		`- }`
	`1516`	`+ pendingAddEntries.add(existsOp);`
`1519`	`1517`	`}`
`1520`	`1518`	`} while (existsOp != null && --pendingSize > 0);`
`1521`	`1519`