feature: Pulsar pool messages support in failure handlers

smallrye-reactive-messaging-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);
     }

smallrye-reactive-messaging-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);
     }

smallrye-reactive-messaging-pulsar/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);
+    }
+
+}

smallrye-reactive-messaging-pulsar/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);
+    }
+
+}