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/KafkaRequestReply.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReply.java
index 651b00267..0bc4ec68c 100644
--- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReply.java
+++ b/smallrye-reactive-messaging-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/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..161ccf6a2 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
@@ -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/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java
index 209b033b6..b3145ba8a 100644
--- a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyTest.java
+++ b/smallrye-reactive-messaging-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);