Add option to disable request reply initial timeout

smallrye · Sep 23, 2024 · 8db3777 · 8db3777
1 parent 03f05cf
commit 8db3777
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 6 deletions.
diff --git a/documentation/src/main/docs/kafka/request-reply.md b/documentation/src/main/docs/kafka/request-reply.md
@@ -94,6 +94,7 @@ If `auto.offset.reset` is `latest`, at wiring time, before any request can take
 finds partitions that the consumer needs to subscribe and waits for their assignment to the consumer.
 The timeout of the initial subscription can be adjusted with `reply.initial-assignment-timeout` which defaults to the `reply.timeout`.
 If this timeout fails, `KafkaRequestReply` will enter an invalid state which will require it to be restarted.
+If set to `-1`, the `KafkaRequestReply` will not wait for the initial assignment of the reply consumer to sent requests.
 
 On other occasions the `KafkaRequestReply#waitForAssignments` method can be used.
 

diff --git a/...ing-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReply.java b/...ing-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReply.java
@@ -76,6 +76,7 @@ public interface KafkaRequestReply<Req, Rep> extends EmitterType {
     /**
      * The config key for the initial assignment timeout.
      * This timeout is used at start when the {@code auto.offset.reset} is set to {@code latest}.
+     * The value {@code -1} disables waiting for initial assignment.
      */
     String REPLY_INITIAL_ASSIGNMENT_TIMEOUT_KEY = "reply.initial-assignment-timeout";
 

diff --git a/...kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java b/...kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java
@@ -7,6 +7,7 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
@@ -104,8 +105,10 @@ public KafkaRequestReplyImpl(EmitterConfiguration config,
         this.replyTopic = consumerConfig.getTopic().orElse(null);
         this.replyPartition = connectorConfig.getOptionalValue(REPLY_PARTITION_KEY, Integer.class).orElse(-1);
         this.replyTimeout = Duration.ofMillis(connectorConfig.getOptionalValue(REPLY_TIMEOUT_KEY, Integer.class).orElse(5000));
-        this.initialAssignmentTimeout = Duration.ofMillis(connectorConfig
-                .getOptionalValue(REPLY_INITIAL_ASSIGNMENT_TIMEOUT_KEY, Integer.class).orElse((int) replyTimeout.toMillis()));
+        int initialAssignmentTimeoutMillis = connectorConfig
+                .getOptionalValue(REPLY_INITIAL_ASSIGNMENT_TIMEOUT_KEY, Integer.class)
+                .orElse((int) replyTimeout.toMillis());
+        this.initialAssignmentTimeout = initialAssignmentTimeoutMillis < 0 ? null : Duration.ofMillis(initialAssignmentTimeoutMillis);
 
         this.autoOffsetReset = consumerConfig.getAutoOffsetReset();
         this.replyCorrelationIdHeader = connectorConfig.getOptionalValue(REPLY_CORRELATION_ID_HEADER_KEY, String.class)
@@ -154,10 +157,11 @@ private Set<TopicPartition> getWaitForPartitions(KafkaConnectorIncomingConfigura
     @Override
     public Flow.Publisher<Message<? extends Req>> getPublisher() {
         return this.publisher
-                .plug(m -> "latest".equals(autoOffsetReset) ? m.onSubscription().call(() -> waitForAssignments()
-                        .ifNoItem()
-                        .after(initialAssignmentTimeout)
-                        .fail())
+                .plug(m -> initialAssignmentTimeout != null && "latest".equals(autoOffsetReset)
+                        ? m.onSubscription().call(() -> waitForAssignments()
+                                .ifNoItem()
+                                .after(initialAssignmentTimeout)
+                                .fail())
                         : m)
                 .onTermination().invoke(this::complete);
     }

diff --git a/...kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java b/...kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java
@@ -479,6 +479,30 @@ void testReplyOffsetResetEarliest() {
                 .extracting(ConsumerRecord::value).containsExactlyInAnyOrder("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
     }
 
+    @Test
+    void testReplyWithoutWaitForAssignment() {
+        addBeans(ReplyServer.class);
+        topic = companion.topics().createAndWait(topic, 3);
+        String replyTopic = topic + "-replies";
+        companion.topics().createAndWait(replyTopic, 3);
+
+        List<String> replies = new CopyOnWriteArrayList<>();
+
+        RequestReplyProducer app = runApplication(config()
+                .withPrefix("mp.messaging.outgoing.request-reply")
+                .with("reply.initial-assignment-timeout", "-1"), RequestReplyProducer.class);
+
+        for (int i = 0; i < 10; i++) {
+            app.requestReply().request(Message.of(i, Metadata.of(OutgoingKafkaRecordMetadata.builder()
+                    .withKey("" + i).build()))).subscribe().with(r -> replies.add(r.getPayload()));
+        }
+        await().untilAsserted(() -> assertThat(replies).hasSize(10));
+        assertThat(replies).containsExactlyInAnyOrder("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
+
+        assertThat(companion.consumeStrings().fromTopics(replyTopic, 10).awaitCompletion())
+                .extracting(ConsumerRecord::value).containsExactlyInAnyOrder("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
+    }
+
     @Test
     void testReplyAssignAndSeekOffset() {
         addBeans(ReplyServer.class);