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/05/20 11:28:31 UTC
(pulsar-client-reactive) branch main updated: Fix negative acknowledgement handling for ReactiveMessagePipeline when message handler throws an exception (#213)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-reactive.git
The following commit(s) were added to refs/heads/main by this push:
new 1d71100 Fix negative acknowledgement handling for ReactiveMessagePipeline when message handler throws an exception (#213)
1d71100 is described below
commit 1d711001915c13363ce0649d156e827e737365b9
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue May 20 14:28:26 2025 +0300
Fix negative acknowledgement handling for ReactiveMessagePipeline when message handler throws an exception (#213)
* Add unit test for negative acknowledgement
* Fix negative acknowledgement handling when messageHandler function throws an exception
* Add javadoc
* Add javadoc comment about ordered processing and negative acknowledgements
---
.../client/api/ReactiveMessagePipelineBuilder.java | 19 ++++++++++++++++++-
.../api/DefaultReactiveMessagePipeline.java | 2 +-
.../client/api/ReactiveMessagePipelineTests.java | 22 ++++++++++++++++++++++
3 files changed, 41 insertions(+), 2 deletions(-)
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineBuilder.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineBuilder.java
index 371b2d4..0b70258 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineBuilder.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineBuilder.java
@@ -37,7 +37,23 @@ import reactor.util.retry.Retry;
public interface ReactiveMessagePipelineBuilder<T> {
/**
- * Sets a handler function that processes messages one-by-one.
+ * <p>
+ * Sets a handler function that processes messages one-by-one. When the message
+ * handler completes successfully, the message will be acknowledged. When the message
+ * handler emits an error, the error logger will be used to log the error and the
+ * message will be negatively acknowledged. If the error logger is not set, a default
+ * error message will be logged at the error level.
+ * </p>
+ * <p>
+ * NOTE: Be aware that negative acknowledgements on ordered subscription types such as
+ * Exclusive, Failover and Key_Shared typically cause failed messages to be sent to
+ * consumers out of their original order. Negative acknowledgements for Key_Shared
+ * subscriptions may also cause message delivery to be blocked on broker versions
+ * before Pulsar 4.0. To maintain ordered message processing, it is recommended to
+ * wrap the message handler with Project Reactor's native retry logic using <a href=
+ * "https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html#retryWhen-reactor.util.retry.Retry-">Mono.retryWhen</a>
+ * to retry processing of each message indefinitely with backoff.
+ * </p>
* @param messageHandler a function that takes a message as input and returns an empty
* Publisher
* @return a builder for the pipeline handling messages one-by-one
@@ -95,6 +111,7 @@ public interface ReactiveMessagePipelineBuilder<T> {
/**
* Sets a function which will be called when the message handler emits an error.
+ * If not set, a default message will be logged at the error level.
* @param errorLogger the error logger function
* @return the pipeline builder instance
*/
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
index 7dc1150..cd94f19 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
@@ -143,7 +143,7 @@ class DefaultReactiveMessagePipeline<T> implements ReactiveMessagePipeline {
}
private Mono<MessageResult<Void>> handleMessage(Message<T> message) {
- return Mono.from(this.messageHandler.apply(message))
+ return Mono.defer(() -> Mono.from(this.messageHandler.apply(message)))
.transform(this::decorateMessageHandler)
.thenReturn(MessageResult.acknowledge(message.getMessageId()))
.onErrorResume((throwable) -> {
diff --git a/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java b/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
index 3e0ddb8..3415481 100644
--- a/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
+++ b/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
@@ -227,6 +227,28 @@ class ReactiveMessagePipelineTests {
}
}
+ @Test
+ void negativeAcknowledgement() throws Exception {
+ int numMessages = 10;
+ TestConsumer testConsumer = new TestConsumer(numMessages);
+ CountDownLatch latch = new CountDownLatch(1);
+ testConsumer.setFinishedCallback(latch::countDown);
+ Function<Message<String>, Publisher<Void>> messageHandler = (message) -> {
+ if (message.getValue().equals("5")) {
+ throw new RuntimeException("Handling message 5 failed");
+ }
+ return Mono.empty();
+ };
+ try (ReactiveMessagePipeline pipeline = testConsumer.messagePipeline().messageHandler(messageHandler).build()) {
+ pipeline.start();
+ assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue();
+ // 9 messages should have been acked
+ assertThat(testConsumer.getAcknowledgedMessages()).hasSize(9);
+ // 1 message should have been nacked
+ assertThat(testConsumer.getNackedMessages()).hasSize(1);
+ }
+ }
+
@Test
void errorLogger() throws Exception {
int numMessages = 10;