Pulsar nack release message only when the message is not used later

smallrye · Apr 15, 2024 · 14672b9 · 14672b9
1 parent 02538ac
commit 14672b9
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 3 deletions.
diff --git a/...saging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarIgnore.java b/...saging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarIgnore.java
@@ -42,7 +42,6 @@ public PulsarIgnore(String channel) {
     public Uni<Void> handle(PulsarIncomingMessage<?> message, Throwable reason, Metadata metadata) {
         log.messageFailureIgnored(channel, reason.getMessage());
         log.messageFailureFullCause(reason);
-        message.unwrap().release();
         return Uni.createFrom().completionStage(message.ack())
                 .emitOn(message::runOnMessageContext);
     }

diff --git a/...essaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarNack.java b/...essaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarNack.java
@@ -45,7 +45,6 @@ public Uni<Void> handle(PulsarIncomingMessage<?> message, Throwable reason, Meta
         consumer.negativeAcknowledge(message.getMessageId());
         log.messageFailureNacked(channel, reason.getMessage());
         log.messageFailureFullCause(reason);
-        message.unwrap().release();
         return Uni.createFrom().voidItem()
                 .emitOn(message::runOnMessageContext);
     }

diff --git a/...ulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarReconsumeLater.java b/...ulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarReconsumeLater.java
@@ -59,7 +59,6 @@ public Uni<Void> handle(PulsarIncomingMessage<?> message, Throwable reason, Meta
 
         log.messageFailureDelayed(channel, delay.toSeconds(), reason.getMessage());
         log.messageFailureFullCause(reason);
-        message.unwrap().release();
         return Uni.createFrom()
                 .completionStage(
                         () -> consumer.reconsumeLaterAsync(message.unwrap(), customProperties, delay.toSeconds(), SECONDS))

diff --git a/...ar/src/test/java/io/smallrye/reactive/messaging/pulsar/ack/PulsarAckPoolMessagesTest.java b/...ar/src/test/java/io/smallrye/reactive/messaging/pulsar/ack/PulsarAckPoolMessagesTest.java
@@ -0,0 +1,12 @@
+package io.smallrye.reactive.messaging.pulsar.ack;
+
+import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
+
+public class PulsarAckPoolMessagesTest extends PulsarAckTest {
+
+    MapBasedConfig config() {
+        return super.config()
+                .with("mp.messaging.incoming.data.poolMessages", true);
+    }
+
+}
diff --git a/...src/test/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarNackPoolMessagesTest.java b/...src/test/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarNackPoolMessagesTest.java
@@ -0,0 +1,12 @@
+package io.smallrye.reactive.messaging.pulsar.fault;
+
+import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
+
+public class PulsarNackPoolMessagesTest extends PulsarNackTest {
+
+    MapBasedConfig config() {
+        return super.config()
+                .with("mp.messaging.incoming.data.poolMessages", true);
+    }
+
+}