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

(cherry picked from commit 420f62e)
(cherry picked from commit 0a0b3ca)

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/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;
@@ -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();
     }
 
 }