From 494019874f69636e70aa18ae99c0a750bc86cd52 Mon Sep 17 00:00:00 2001 From: Ozan Gunalp Date: Thu, 19 Sep 2024 14:34:45 +0200 Subject: [PATCH] Add option to disable request reply initial timeout --- documentation/src/main/docs/kafka/request-reply.md | 1 + .../messaging/kafka/reply/KafkaRequestReplyImpl.java | 9 ++++++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/documentation/src/main/docs/kafka/request-reply.md b/documentation/src/main/docs/kafka/request-reply.md index 8cf79f719..2ef09781d 100644 --- 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/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java index 13c001f11..d0cdc7770 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java +++ b/smallrye-reactive-messaging-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 getWaitForPartitions(KafkaConnectorIncomingConfigura @Override public Flow.Publisher> 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())