Fix ack coordination for empty publishers in POST_PROCESSING strategy of payload -> stream signatures

Fixes #3232

Author: ozangunalp

smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java

@@ -281,6 +281,25 @@ private void processMethodReturningAProcessorOfPayloads() {
         };
     }
 
+    private Multi<? extends Message<?>> handleMultiAckCoordination(Message<?> message, Flow.Publisher<?> pubFlow) {
+        // POST_PROCESSING must not be used when returning an infinite stream
+        AcknowledgementCoordinator coordinator = new AcknowledgementCoordinator(message);
+        return MultiUtils.publisher(pubFlow)
+                .onItem()
+                .transform(payload -> coordinator.track(payloadToMessage(payload, message.getMetadata())))
+                .onTermination().call((throwable, cancelled) -> {
+                    if (coordinator.getTrackedCount() == 0) {
+                        // If the publisher terminates without emitting any item, we need to ack the original message.
+                        if (throwable == null && !cancelled) {
+                            return Uni.createFrom().completionStage(message.ack());
+                        }
+                        return Uni.createFrom().completionStage(message.nack(throwable));
+                    } else {
+                        return Uni.createFrom().voidItem();
+                    }
+                });
+    }
+
     /**
      * {@code PublisherBuilder<O> method(I payload)}
      */
@@ -291,11 +310,7 @@ private void processMethodReturningAPublisherBuilderOfPayloadsAndConsumingPayloa
                 if (isPostAck()) {
                     try {
                         PublisherBuilder<?> pb = invoke(getArguments(message));
-                        // POST_PROCESSING must not be used when returning an infinite stream
-                        AcknowledgementCoordinator coordinator = new AcknowledgementCoordinator(message);
-                        return MultiUtils.publisher(AdaptersToFlow.publisher(pb.buildRs()))
-                                .onItem()
-                                .transform(payload -> coordinator.track(payloadToMessage(payload, message.getMetadata())));
+                        return handleMultiAckCoordination(message, AdaptersToFlow.publisher(pb.buildRs()));
                     } catch (Throwable t) {
                         return handlePostInvocation(message, t);
                     }
@@ -318,11 +333,7 @@ private void processMethodReturningAReactiveStreamsPublisherOfPayloadsAndConsumi
                 if (isPostAck()) {
                     try {
                         Publisher<?> pub = invoke(getArguments(message));
-                        // POST_PROCESSING must not be used when returning an infinite stream
-                        AcknowledgementCoordinator coordinator = new AcknowledgementCoordinator(message);
-                        return MultiUtils.publisher(AdaptersToFlow.publisher(pub))
-                                .onItem()
-                                .transform(payload -> coordinator.track(payloadToMessage(payload, message.getMetadata())));
+                        return handleMultiAckCoordination(message, AdaptersToFlow.publisher(pub));
                     } catch (Throwable t) {
                         return handlePostInvocation(message, t);
                     }
@@ -345,12 +356,7 @@ private void processMethodReturningAPublisherOfPayloadsAndConsumingPayloads() {
                 if (isPostAck()) {
                     try {
                         Flow.Publisher<?> pub = invoke(getArguments(message));
-                        // POST_PROCESSING must not be used when returning an infinite stream
-                        AcknowledgementCoordinator coordinator = new AcknowledgementCoordinator(message);
-                        return MultiUtils.publisher(pub)
-                                .onItem()
-                                .transform(payload -> coordinator.track(payloadToMessage(payload, message.getMetadata())));
-
+                        return handleMultiAckCoordination(message, pub);
                     } catch (Throwable t) {
                         return handlePostInvocation(message, t);
                     }

smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/helpers/AcknowledgementCoordinator.java

@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.eclipse.microprofile.reactive.messaging.Message;
@@ -21,6 +22,7 @@ public class AcknowledgementCoordinator {
     private final Message<?> input;
 
     private volatile boolean done;
+    private final AtomicInteger trackedCount = new AtomicInteger();
     private final List<Tracker> tracked = new ArrayList<>();
 
     private final ReentrantLock lock = new ReentrantLock();
@@ -33,6 +35,7 @@ public Message<?> track(Message<?> msg) {
         lock.lock();
         try {
             Tracker tracker = new Tracker();
+            trackedCount.getAndIncrement();
             tracked.add(tracker);
             return msg
                     .withAck(() -> {
@@ -48,6 +51,10 @@ public Message<?> track(Message<?> msg) {
         }
     }
 
+    public int getTrackedCount() {
+        return trackedCount.get();
+    }
+
     private void onAck(Tracker id) {
         lock.lock();
         try {

smallrye-reactive-messaging-provider/src/test/java/io/smallrye/reactive/messaging/acknowledgement/AsynchronousPayloadProcessorAckTest.java

@@ -3,9 +3,11 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.*;
+import java.util.stream.Stream;
 
 import jakarta.enterprise.context.ApplicationScoped;
 
@@ -130,6 +132,71 @@ public void testThatMessagesAreNackedAfterFailingProcessingOfPayloadUni() throws
         assertThat(throwables).hasSize(2);
     }
 
+    @ApplicationScoped
+    public static class StringToStringMultiProcessorWithPostProcessingAck {
+
+        @Incoming("data")
+        @Outgoing("out")
+        @Acknowledgment(Acknowledgment.Strategy.POST_PROCESSING)
+        public Multi<String> process(String incomingData) {
+            return Multi.createFrom().items(getWordsStream(incomingData));
+        }
+
+        private Stream<String> getWordsStream(String sentence) {
+            return Arrays.stream(sentence.split(" "));
+        }
+    }
+
+    @Test
+    public void testAckingMessagesForStringToStringMultiProcessorWithPostProcessingAck() throws InterruptedException {
+        addBeanClass(StringToStringMultiProcessorWithPostProcessingAck.class);
+        initialize();
+        Emitter<String> emitter = get(EmitterBean.class).emitter();
+        Sink sink = get(Sink.class);
+
+        Set<String> acked = new ConcurrentHashSet<>();
+        Set<String> nacked = new ConcurrentHashSet<>();
+
+        run(acked, nacked, emitter);
+
+        await().until(() -> sink.list().size() == 10);
+        assertThat(acked).hasSize(10);
+        assertThat(nacked).hasSize(0);
+    }
+
+    @ApplicationScoped
+    public static class StringToEmptyStringMultiProcessorWithPostProcessingAck {
+
+        @Incoming("data")
+        @Outgoing("out")
+        @Acknowledgment(Acknowledgment.Strategy.POST_PROCESSING)
+        public Multi<String> process(String incomingData) {
+            return Multi.createFrom().items(getWordsStream(incomingData));
+        }
+
+        private Stream<String> getWordsStream(String sentence) {
+            return Stream.empty();
+        }
+    }
+
+    @Test
+    public void testAckingMessagesForStringToEmptyStringMultiProcessorWithPostProcessingAck() throws InterruptedException {
+        addBeanClass(StringToEmptyStringMultiProcessorWithPostProcessingAck.class);
+        initialize();
+        Emitter<String> emitter = get(EmitterBean.class).emitter();
+        Sink sink = get(Sink.class);
+
+        Set<String> acked = new ConcurrentHashSet<>();
+        Set<String> nacked = new ConcurrentHashSet<>();
+
+        run(acked, nacked, emitter);
+
+        // All items are skipped
+        await().pollDelay(2, TimeUnit.SECONDS).untilAsserted(() -> assertThat(sink.list()).isEmpty());
+        assertThat(acked).hasSize(10);
+        assertThat(nacked).hasSize(0);
+    }
+
     private List<Throwable> run(Set<String> acked, Set<String> nacked, Emitter<String> emitter)
             throws InterruptedException {
         List<Throwable> reasons = new CopyOnWriteArrayList<>();