Add option to disable request reply initial timeout

smallrye · Sep 19, 2024 · 4940198 · 4940198
1 parent 03f05cf
commit 4940198
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 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/...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
@@ -104,8 +104,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()));
+        this.initialAssignmentTimeout = connectorConfig
+                .getOptionalValue(REPLY_INITIAL_ASSIGNMENT_TIMEOUT_KEY, Integer.class)
+                .map(timeout -> timeout < 0 ? null : Duration.ofMillis(timeout))
+                .orElse(replyTimeout);
 
         this.autoOffsetReset = consumerConfig.getAutoOffsetReset();
         this.replyCorrelationIdHeader = connectorConfig.getOptionalValue(REPLY_CORRELATION_ID_HEADER_KEY, String.class)
@@ -154,7 +156,8 @@ 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()
+                .plug(m -> initialAssignmentTimeout != null && "latest".equals(autoOffsetReset)
+                        ? m.onSubscription().call(() -> waitForAssignments()
                         .ifNoItem()
                         .after(initialAssignmentTimeout)
                         .fail())