You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/01/07 17:26:06 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Remove failed OpAddEntry from pendingAddEntries (#23817)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new ef15ce3ede7 [fix][broker] Remove failed OpAddEntry from pendingAddEntries (#23817)
ef15ce3ede7 is described below
commit ef15ce3ede7d1f218005f81af70e825dfb50483b
Author: ran <rg...@streamnative.io>
AuthorDate: Wed Jan 8 00:34:45 2025 +0800
[fix][broker] Remove failed OpAddEntry from pendingAddEntries (#23817)
(cherry picked from commit 420f62eef20aaef49ae404a683f2d5466ccfdec3)
---
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 1 +
.../impl/MangedLedgerInterceptorImpl2Test.java | 8 ++---
....java => ManagedLedgerInterceptorImplTest.java} | 40 ++++++++++++++++------
3 files changed, 35 insertions(+), 14 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 2c00fd2cdf7..e0d35ce4e91 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -143,6 +143,7 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable, Managed
payloadProcessorHandle = ml.getManagedLedgerInterceptor()
.processPayloadBeforeLedgerWrite(this.getCtx(), 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));
diff --git a/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/MangedLedgerInterceptorImpl2Test.java b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/MangedLedgerInterceptorImpl2Test.java
index 1be66a7f9d8..ec05f4dc381 100644
--- a/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/MangedLedgerInterceptorImpl2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/MangedLedgerInterceptorImpl2Test.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.ManagedLedgerConfig;
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 class MangedLedgerInterceptorImpl2Test extends MockedBookKeeperTestCase {
switchLedgerManually(ledger);
// verify.
- assertEquals(currentLedgerSize, MangedLedgerInterceptorImplTest.calculatePreciseSize(ledger));
+ assertEquals(currentLedgerSize, ManagedLedgerInterceptorImplTest.calculatePreciseSize(ledger));
// cleanup.
cursor.close();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
similarity index 93%
rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
index 26b2d52c194..3866130c64b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.intercept;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -33,6 +32,7 @@ import java.util.List;
import java.util.Set;
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;
@@ -59,8 +59,8 @@ import org.testng.Assert;
import org.testng.annotations.Test;
@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
@@ -446,26 +446,33 @@ public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
return new Processor() {
@Override
public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
- throw new RuntimeException(failureMsg);
+ Commands.skipBrokerEntryMetadataIfExist(inputPayload);
+ 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
@@ -474,10 +481,23 @@ public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
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();
}
}