feature: Pulsar pool messages support in failure handlers

michalcukierman · Apr 15, 2024 · 02538ac · 02538ac
1 parent b38a33b
commit 02538ac
Show file tree

Hide file tree

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

diff --git a/...ging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarFailStop.java b/...ging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarFailStop.java
@@ -45,6 +45,7 @@ public Uni<Void> handle(PulsarIncomingMessage<?> message, Throwable reason, Meta
         // We don't commit, we just fail and stop the client.
         log.messageFailureFailStop(channel);
         reportFailure.accept(reason, true);
+        message.unwrap().release();
         return Uni.createFrom().<Void> failure(reason)
                 .emitOn(message::runOnMessageContext);
     }

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,6 +42,7 @@ 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,6 +45,7 @@ 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,6 +59,7 @@ 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))