[fix][broker] Remove failed OpAddEntry from pendingAddEntries (#23817)

(cherry picked from commit 420f62e)

Author: gaoran10

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

@@ -142,6 +142,7 @@ public void initiate() {
                     payloadProcessorHandle = ml.getManagedLedgerInterceptor()
                             .processPayloadBeforeLedgerWrite(this, duplicateBuffer);
                 } catch (Exception e) {
+                    ml.pendingAddEntries.remove(this);
                     ReferenceCountUtil.safeRelease(duplicateBuffer);
                     log.error("[{}] Error processing payload before ledger write", ml.getName(), e);
                     this.failed(new ManagedLedgerException.ManagedLedgerInterceptException(e));

pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/MangedLedgerInterceptorImplTest2.java

@@ -19,7 +19,7 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import static org.testng.Assert.assertEquals;
-import static org.apache.pulsar.broker.intercept.MangedLedgerInterceptorImplTest.TestPayloadProcessor;
+import static org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImplTest.TestPayloadProcessor;
 import java.util.HashSet;
 import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
@@ -28,13 +28,13 @@
 import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
 import org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImpl;
-import org.apache.pulsar.broker.intercept.MangedLedgerInterceptorImplTest;
+import org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImplTest;
 import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
 import org.awaitility.Awaitility;
 import org.testng.annotations.Test;
 
 /***
- * Differ to {@link MangedLedgerInterceptorImplTest}, this test can call {@link ManagedLedgerImpl}'s methods modified
+ * Differ to {@link ManagedLedgerInterceptorImplTest}, this test can call {@link ManagedLedgerImpl}'s methods modified
  * by "default".
  */
 @Slf4j
@@ -73,7 +73,7 @@ public void testCurrentLedgerSizeCorrectIfHasInterceptor() throws Exception {
         switchLedgerManually(ledger);
 
         // verify.
-        assertEquals(currentLedgerSize, MangedLedgerInterceptorImplTest.calculatePreciseSize(ledger));
+        assertEquals(currentLedgerSize, ManagedLedgerInterceptorImplTest.calculatePreciseSize(ledger));
 
         // cleanup.
         cursor.close();

…ept/MangedLedgerInterceptorImplTest.java …pt/ManagedLedgerInterceptorImplTest.javapulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java renamed to pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java renamed to pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java

@@ -18,14 +18,14 @@
  */
 package org.apache.pulsar.broker.intercept;
 
-import static org.testng.Assert.fail;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 import lombok.Cleanup;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
@@ -63,8 +63,8 @@
 import static org.testng.Assert.assertNotNull;
 
 @Test(groups = "broker")
-public class MangedLedgerInterceptorImplTest  extends MockedBookKeeperTestCase {
-    private static final Logger log = LoggerFactory.getLogger(MangedLedgerInterceptorImplTest.class);
+public class ManagedLedgerInterceptorImplTest  extends MockedBookKeeperTestCase {
+    private static final Logger log = LoggerFactory.getLogger(ManagedLedgerInterceptorImplTest.class);
 
     public static class TestPayloadProcessor implements ManagedLedgerPayloadProcessor {
         @Override
@@ -451,26 +451,32 @@ public Processor inputProcessor() {
                 return new Processor() {
                     @Override
                     public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
-                        throw new RuntimeException(failureMsg);
+                        if (inputPayload.readBoolean()) {
+                            throw new RuntimeException(failureMsg);
+                        }
+                        return inputPayload;
                     }
 
                     @Override
                     public void release(ByteBuf processedPayload) {
                         // no-op
-                        fail("the release method can't be reached");
                     }
                 };
             }
         })));
 
         var ledger = factory.open("testManagedLedgerPayloadProcessorFailure", config);
-        var countDownLatch = new CountDownLatch(1);
+        int count = 10;
+        var countDownLatch = new CountDownLatch(count);
+        var successCount = new AtomicInteger(0);
         var expectedException = new ArrayList<Exception>();
-        ledger.asyncAddEntry("test".getBytes(), 1, 1, new AsyncCallbacks.AddEntryCallback() {
+
+        var addEntryCallback = new AsyncCallbacks.AddEntryCallback() {
             @Override
             public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                 entryData.release();
                 countDownLatch.countDown();
+                successCount.incrementAndGet();
             }
 
             @Override
@@ -479,10 +485,23 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
                 expectedException.add(exception);
                 countDownLatch.countDown();
             }
-        }, null);
+        };
+
+        for (int i = 0; i < count; i++) {
+            if (i % 2 == 0) {
+                ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(true), addEntryCallback, null);
+            } else {
+                ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(false), addEntryCallback, null);
+            }
+        }
+
         countDownLatch.await();
-        assertEquals(expectedException.size(), 1);
-        assertEquals(expectedException.get(0).getCause().getMessage(), failureMsg);
+        assertEquals(expectedException.size(), count / 2);
+        assertEquals(successCount.get(), count / 2);
+        for (Exception e : expectedException) {
+            assertEquals(e.getCause().getMessage(), failureMsg);
+        }
+        ledger.close();
     }
 
 }