Custom timeout exception for Kafka request reply

Author: ozangunalp

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java

@@ -236,10 +236,9 @@ public Multi<Message<Rep>> requestMulti(Message<Req> request) {
                                     replyTopic,
                                     replyPartition,
                                     (MultiEmitter<Message<Rep>>) emitter));
-                })
-                        .ifNoItem().after(replyTimeout).fail())
-                .onTermination().invoke(() -> pendingReplies.remove(correlationId))
-                .onItem().transformToUniAndMerge(m -> {
+                }))
+                .ifNoItem().after(replyTimeout).failWith(() -> new KafkaRequestReplyTimeoutException(correlationId))
+                .onItem().transformToUniAndConcatenate(m -> {
                     if (replyFailureHandler != null) {
                         Throwable failure = replyFailureHandler.handleReply((KafkaRecord<?, ?>) m);
                         if (failure != null) {
@@ -248,6 +247,7 @@ public Multi<Message<Rep>> requestMulti(Message<Req> request) {
                     }
                     return Uni.createFrom().item(m);
                 })
+                .onTermination().invoke(() -> pendingReplies.remove(correlationId))
                 .plug(multi -> replyConverter != null ? multi.map(f -> replyConverter.apply(f)) : multi);
     }
 

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTimeoutException.java

@@ -0,0 +1,12 @@
+package io.smallrye.reactive.messaging.kafka.reply;
+
+/**
+ * Exception thrown when a reply is not received within the configured timeout.
+ */
+public class KafkaRequestReplyTimeoutException extends RuntimeException {
+
+    public KafkaRequestReplyTimeoutException(CorrelationId correlationId) {
+        super("Timeout waiting for a reply for request with correlation ID: " + correlationId);
+    }
+
+}

smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java

@@ -628,7 +628,7 @@ void testReplyTimeout() {
 
         app.requestReply().request(1)
                 .subscribe().withSubscriber(UniAssertSubscriber.create())
-                .awaitFailure().assertFailedWith(TimeoutException.class);
+                .awaitFailure().assertFailedWith(KafkaRequestReplyTimeoutException.class);
     }
 
     @Test