[fix][broker] Fix duplicate increment of ADD_OP_COUNT_UPDATER in OpAddEntry (#24506)

Co-authored-by: crossoverJie <[email protected]>

Author: 3pacccccc

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java

@@ -135,7 +135,6 @@ public void initiate() {
             ByteBuf duplicateBuffer = data.retainedDuplicate();
 
             // internally asyncAddEntry() will take the ownership of the buffer and release it at the end
-            addOpCount = ManagedLedgerImpl.ADD_OP_COUNT_UPDATER.incrementAndGet(ml);
             lastInitTime = System.nanoTime();
             if (ml.getManagedLedgerInterceptor() != null) {
                 long originalDataLen = data.readableBytes();

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -3318,6 +3318,25 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
         setFieldValue(ManagedLedgerImpl.class, ledger, "currentLedger", null);
     }
 
+    @Test
+    public void testAddOpCountWithMessageAdd() throws Exception {
+        ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig();
+        config.setMaxCacheSize(0);
+
+        @Cleanup("shutdown")
+        ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, config);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("my_test_ledger");
+
+        for (int i = 0; i < 10; i++) {
+            OpAddEntry op = OpAddEntry.createNoRetainBuffer(ledger,
+                    ByteBufAllocator.DEFAULT.buffer(128), null, null, new AtomicBoolean());
+            ledger.internalAsyncAddEntry(op);
+            long addOpCount = ManagedLedgerImpl.ADD_OP_COUNT_UPDATER.get(ledger);
+            Assert.assertEquals(i + 1, addOpCount);
+        }
+    }
+
+
     @Test
     public void avoidUseSameOpAddEntryBetweenDifferentLedger() throws Exception {
         ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig();