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())