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/07/24 18:04:56 UTC
(pulsar) branch branch-3.0 updated: [fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 4f156b5661b [fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
4f156b5661b is described below
commit 4f156b5661b842c1ff025a9ec6c555a426dd4ec5
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Thu Jul 24 18:16:17 2025 +0800
[fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
Co-authored-by: Yunze Xu <xy...@163.com>
(cherry picked from commit a14794a2a73142f5cbd4fe13b8fca18532ebd001)
---
.../apache/bookkeeper/mledger/ManagedLedger.java | 17 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 11 ++
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 18 ++-
.../broker/service/persistent/PersistentTopic.java | 1 +
.../broker/ManagedLedgerPayloadProcessor0.java | 57 +++++++++
.../broker/PublishWithMLPayloadProcessorTest.java | 131 +++++++++++++++++++++
.../ManagedLedgerInterceptorImplTest.java | 45 ++++---
7 files changed, 261 insertions(+), 19 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index 52e5d3366dd..aab21ba7cfd 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -417,6 +417,23 @@ public interface ManagedLedger {
*/
long getOffloadedSize();
+ /**
+ * Resets the exception thrown by the PayloadProcessor during an add entry operation to null.
+ * <p>
+ * **Context:** When an add entry operation fails due to an interceptor, all subsequent incoming add entry
+ * operations will also fail. This behavior ensures message ordering and consistency.
+ * <p>
+ * **Important:** This method MUST only be called after all pending add operations are fully completed
+ * (e.g., after a Topic is unfenced). Calling it prematurely will prevent the Managed Ledger (ML)
+ * from being able to write indefinitely.
+ * <p>
+ * **Implementation Note:** Downstream projects that support the ML PayloadProcessor should implement
+ * this method. Otherwise, do not implement it.
+ */
+ default void unfenceForInterceptorException() {
+ // Default implementation does nothing
+ }
+
/**
* Get last offloaded ledgerId. If no offloaded yet, it returns 0.
*
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
index 4392fbc1f26..54fa3dbe8e4 100644
--- 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
@@ -231,6 +231,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private volatile long lastOffloadSuccessTimestamp = 0;
private volatile long lastOffloadFailureTimestamp = 0;
+ protected volatile ManagedLedgerException interceptorException = null;
+
private int minBacklogCursorsForCaching = 0;
private int minBacklogEntriesForCaching = 1000;
private int maxBacklogBetweenCursorsForCaching = 1000;
@@ -4316,6 +4318,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return offloadedSize;
}
+ @Override
+ public void unfenceForInterceptorException() {
+ this.interceptorException = null;
+ }
+
+ protected void fenceForInterceptorException(ManagedLedgerException e) {
+ this.interceptorException = e;
+ }
+
@Override
public long getLastOffloadedLedgerId() {
return lastOffloadLedgerId;
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 3d0c33851b2..27f540c8642 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
@@ -131,8 +131,16 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
public void initiate() {
if (STATE_UPDATER.compareAndSet(OpAddEntry.this, State.OPEN, State.INITIATED)) {
- ByteBuf duplicateBuffer = data.retainedDuplicate();
+ // Fail the add operation if the managed ledger is in a state that prevents adding entries.
+ ManagedLedgerException exbw = ml.interceptorException;
+ if (exbw != null) {
+ ml.pendingAddEntries.remove(this);
+ this.failed(exbw);
+ // Don't recycle the object here, see: https://lists.apache.org/thread/po08w0tkhc7q8gc5khpdft6stxnr1v2y
+ return;
+ }
+ ByteBuf duplicateBuffer = data.retainedDuplicate();
// internally asyncAddEntry() will take the ownership of the buffer and release it at the end
lastInitTime = System.nanoTime();
if (ml.getManagedLedgerInterceptor() != null) {
@@ -141,10 +149,14 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
payloadProcessorHandle = ml.getManagedLedgerInterceptor()
.processPayloadBeforeLedgerWrite(this, duplicateBuffer);
} catch (Exception e) {
+ ManagedLedgerException mle = new ManagedLedgerException.ManagedLedgerInterceptException(e);
+ ml.fenceForInterceptorException(mle);
ml.pendingAddEntries.remove(this);
ReferenceCountUtil.safeRelease(duplicateBuffer);
log.error("[{}] Error processing payload before ledger write", ml.getName(), e);
- this.failed(new ManagedLedgerException.ManagedLedgerInterceptException(e));
+ this.failed(mle);
+ // Don't recycle the object here
+ // see: https://lists.apache.org/thread/po08w0tkhc7q8gc5khpdft6stxnr1v2y
return;
}
if (payloadProcessorHandle != null) {
@@ -169,7 +181,7 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
//Use entryId in PublishContext and call addComplete directly.
this.addComplete(BKException.Code.OK, ledger, ((Position) ctx).getEntryId(), addOpCount);
} else {
- log.warn("[{}] initiate with unexpected state {}, expect OPEN state.", ml.getName(), state);
+ log.warn("[{}] initiateShadowWrite with unexpected state {}, expect OPEN state.", ml.getName(), state);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index c5a76ddf519..3065ce3f70e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -3969,6 +3969,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private synchronized void unfence() {
isFenced = false;
+ ledger.unfenceForInterceptorException();
cancelFencedTopicMonitoringTask();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java
new file mode 100644
index 00000000000..e48a0c7e3bb
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
+
+@SuppressWarnings("unused") // Used by PublishWithMLPayloadProcessorTest
+public class ManagedLedgerPayloadProcessor0 implements ManagedLedgerPayloadProcessor {
+
+ private final AtomicInteger counter = new AtomicInteger(0);
+ private final int failAt = 4;
+
+ @Override
+ public Processor inputProcessor() {
+ return new Processor() {
+ @Override
+ public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ if (counter.incrementAndGet() == failAt) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ throw new RuntimeException("Failed to process input payload");
+ }
+ byte[] bytes = new byte[inputPayload.readableBytes()];
+ inputPayload.readBytes(bytes);
+ inputPayload.release();
+ return Unpooled.wrappedBuffer(bytes, 0, bytes.length).retainedDuplicate();
+ }
+
+ @Override
+ public void release(ByteBuf processedPayload) {
+ processedPayload.release();
+ }
+ };
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java
new file mode 100644
index 00000000000..cafd220b6a6
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class PublishWithMLPayloadProcessorTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setBrokerEntryPayloadProcessors(
+ Collections.singleton(ManagedLedgerPayloadProcessor0.class.getName()));
+ super.internalSetup();
+ super.producerBaseSetup();
+ admin.tenants().createTenant("my-test-tenant", createDefaultTenantInfo());
+ admin.namespaces().createNamespace("my-test-tenant/my-test-ns");
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+
+ @Test(timeOut = 30_000)
+ public void testPublishWithoutDeduplication() throws Exception {
+ String topic = "persistent://my-test-tenant/my-test-ns/testPublishWithoutDeduplication";
+ admin.topics().createNonPartitionedTopic(topic);
+ try {
+ admin.topicPolicies().setDeduplicationStatus(topic, false);
+ publishAndVerify(topic, false);
+ } finally {
+ admin.topics().delete(topic);
+ }
+ }
+
+ @Test(timeOut = 30_000)
+ public void testPublishWithDeduplication() throws Exception {
+ String topic = "persistent://my-test-tenant/my-test-ns/testPublishWithDeduplication";
+ admin.topics().createNonPartitionedTopic(topic);
+ try {
+ admin.topicPolicies().setDeduplicationStatus(topic, true);
+ publishAndVerify(topic, true);
+ } finally {
+ admin.topics().delete(topic);
+ }
+ }
+
+
+ private void publishAndVerify(String topic, boolean enableDeduplication) throws Exception {
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+
+ CountDownLatch latch = new CountDownLatch(10);
+ for (int i = 0; i < 10; i++) {
+ producer.sendAsync("message-" + i).whenComplete((ignored, e) -> {
+ if (e != null) {
+ log.error("Failed to publish message", e);
+ }
+ latch.countDown();
+ });
+ }
+
+ latch.await();
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic).subscriptionName("my-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+
+ List<String> messageContents = new ArrayList<>();
+
+ int messageCount = 0;
+ for (;;) {
+ var msg = consumer.receive(3, TimeUnit.SECONDS);
+ if (msg == null) {
+ break;
+ }
+ String content = msg.getValue();
+ messageCount++;
+ if (!messageContents.contains(content)) {
+ messageContents.add(content);
+ }
+ consumer.acknowledge(msg);
+ }
+
+ if (enableDeduplication) {
+ Assert.assertEquals(messageCount, 10, "Expected 10 total messages");
+ }
+ Assert.assertEquals(messageContents.size(), 10);
+
+ for (int i = 0; i < 10; i++) {
+ Assert.assertEquals(messageContents.get(i), "message-" + i);
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
index 455dcbbf413..a2d033bcad9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
@@ -455,6 +455,7 @@ public class ManagedLedgerInterceptorImplTest extends MockedBookKeeperTestCase
return new Processor() {
@Override
public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ log.info("testManagedLedgerPayloadInputProcessorFailure.process");
if (inputPayload.readBoolean()) {
throw new RuntimeException(failureMsg);
}
@@ -475,26 +476,38 @@ public class ManagedLedgerInterceptorImplTest extends MockedBookKeeperTestCase
var successCount = new AtomicInteger(0);
var expectedException = new ArrayList<Exception>();
- var addEntryCallback = new AsyncCallbacks.AddEntryCallback() {
- @Override
- public void addComplete(Position position, ByteBuf entryData, Object ctx) {
- successCount.incrementAndGet();
- countDownLatch.countDown();
- }
-
- @Override
- public void addFailed(ManagedLedgerException exception, Object ctx) {
- // expected
- expectedException.add(exception);
- countDownLatch.countDown();
- }
- };
+ ByteBuf shouldFail = Unpooled.buffer().writeBoolean(true);
+ ByteBuf shouldSucceed = Unpooled.buffer().writeBoolean(false);
+ byte[] shouldFailBytes = new byte[shouldFail.readableBytes()];
+ shouldFail.readBytes(shouldFailBytes);
+ byte[] shouldSucceedBytes = new byte[shouldSucceed.readableBytes()];
+ shouldSucceed.readBytes(shouldSucceedBytes);
+ shouldSucceed.release();
+ shouldFail.release();
for (int i = 0; i < count; i++) {
if (i % 2 == 0) {
- ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(true), addEntryCallback, null);
+ try {
+ ledger.addEntry(shouldFailBytes);
+ successCount.incrementAndGet();
+ countDownLatch.countDown();
+ } catch (Exception t) {
+ expectedException.add(t);
+ countDownLatch.countDown();
+ } finally {
+ ledger.unfenceForInterceptorException();
+ }
} else {
- ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(false), addEntryCallback, null);
+ try {
+ ledger.addEntry(shouldSucceedBytes);
+ successCount.incrementAndGet();
+ countDownLatch.countDown();
+ } catch (Exception t) {
+ expectedException.add(t);
+ countDownLatch.countDown();
+ } finally {
+ ledger.unfenceForInterceptorException();
+ }
}
}