diff --git a/api/src/main/java/io/smallrye/reactive/messaging/PausableChannel.java b/api/src/main/java/io/smallrye/reactive/messaging/PausableChannel.java index 7ad674f953..b8b35b3fbc 100644 --- a/api/src/main/java/io/smallrye/reactive/messaging/PausableChannel.java +++ b/api/src/main/java/io/smallrye/reactive/messaging/PausableChannel.java @@ -21,4 +21,11 @@ public interface PausableChannel { * Resumes the channel. */ void resume(); + + /** + * Clears the buffer of the channel, discarding any items that have been + * requested from upstream but not yet delivered to the consumer. + */ + default void clearBuffer() { + } } diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/KafkaConsumerRebalanceListener.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/KafkaConsumerRebalanceListener.java index f8b656c272..5ac95df22d 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/KafkaConsumerRebalanceListener.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/KafkaConsumerRebalanceListener.java @@ -142,4 +142,23 @@ default void onPartitionsLost(Consumer consumer, Collection + * This callback is invoked after the seek has taken effect. Calling + * {@link Consumer#position(org.apache.kafka.common.TopicPartition)} on the provided consumer + * will return the new position. + *

+ * This allows listeners to reset any per-partition state that depends on the consumer's position. + *

+ * IMPORTANT: The behavior must be blocking. Callback invoked from the polling thread. + * + * @param consumer the underlying Kafka consumer + * @param partitions the partitions that were seeked + */ + default void onPartitionsSeeked(Consumer consumer, Collection partitions) { + + } + } diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandler.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandler.java index 652b358d97..e7aaa414f6 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandler.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandler.java @@ -94,6 +94,16 @@ default void partitionsRevoked(Collection partitions) { // Do nothing by default. } + /** + * Called when the consumer position is explicitly changed via seek operations, + * allowing handlers to reset per-partition tracking state. + * + * @param partitions seeked partitions + */ + default void partitionsSeeked(Collection partitions) { + // Do nothing by default. + } + /** * Handle message acknowledgment * diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaLatestCommit.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaLatestCommit.java index 035c1c183a..403881f5ee 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaLatestCommit.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaLatestCommit.java @@ -2,6 +2,7 @@ import static io.smallrye.reactive.messaging.kafka.i18n.KafkaLogging.log; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.function.BiConsumer; @@ -34,7 +35,7 @@ */ public class KafkaLatestCommit extends ContextHolder implements KafkaCommitHandler { - private KafkaConsumer consumer; + private final KafkaConsumer consumer; /** * Stores the last offset for each topic/partition. @@ -63,6 +64,21 @@ public KafkaLatestCommit(Vertx vertx, KafkaConnectorIncomingConfiguration config this.consumer = consumer; } + @Override + public void partitionsRevoked(Collection partitions) { + runOnContextAndAwait(() -> { + for (TopicPartition partition : partitions) { + offsets.remove(partition); + } + return null; + }); + } + + @Override + public void partitionsSeeked(Collection partitions) { + partitionsRevoked(partitions); + } + @Override public Uni handle(IncomingKafkaRecord record) { runOnContext(() -> { diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaThrottledLatestProcessedCommit.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaThrottledLatestProcessedCommit.java index c05448a2af..9b290d27c5 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaThrottledLatestProcessedCommit.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaThrottledLatestProcessedCommit.java @@ -56,6 +56,7 @@ public class KafkaThrottledLatestProcessedCommit extends ContextHolder implements KafkaCommitHandler { private final Map offsetStores = new HashMap<>(); + private final Set seekedPartitions = new HashSet<>(); private final String groupId; private final KafkaConsumer consumer; @@ -146,6 +147,7 @@ public void partitionsRevoked(Collection partitions) { Map toCommit = new HashMap<>(); for (TopicPartition partition : new HashSet<>(offsetStores.keySet())) { if (!assignments.contains(partition)) { // revoked partition - remove and compute last commit + seekedPartitions.remove(partition); OffsetStore store = offsetStores.remove(partition); if (store != null) { @@ -174,6 +176,17 @@ public void partitionsRevoked(Collection partitions) { } } + @Override + public void partitionsSeeked(Collection partitions) { + runOnContextAndAwait(() -> { + for (TopicPartition partition : partitions) { + offsetStores.remove(partition); + seekedPartitions.add(partition); + } + return null; + }); + } + /** * Cancel the existing timer. * Must be called from the event loop. @@ -209,15 +222,24 @@ public Uni> received(IncomingKafkaRecord OffsetStore offsetStore = offsetStores.get(recordsTopicPartition); Uni uni; if (offsetStore == null) { - uni = consumer.committed(recordsTopicPartition) - .emitOn(this::runOnContext) // Switch back to event loop - .onItem().transform(offsets -> { - OffsetAndMetadata lastCommitted = offsets.get(recordsTopicPartition); - OffsetStore store = new OffsetStore(recordsTopicPartition, unprocessedRecordMaxAge, - lastCommitted == null ? -1 : lastCommitted.offset() - 1); - offsetStores.put(recordsTopicPartition, store); - return store; - }); + if (seekedPartitions.remove(recordsTopicPartition)) { + uni = Uni.createFrom().item(() -> { + OffsetStore store = new OffsetStore(recordsTopicPartition, unprocessedRecordMaxAge, -1); + offsetStores.put(recordsTopicPartition, store); + return store; + }).emitOn(this::runOnContext); + } else { + uni = consumer.committed(recordsTopicPartition) + .emitOn(this::runOnContext) // Switch back to event loop + .onFailure().recoverWithItem(Collections::emptyMap) + .onItem().transform(offsets -> { + OffsetAndMetadata lastCommitted = offsets.get(recordsTopicPartition); + OffsetStore store = new OffsetStore(recordsTopicPartition, unprocessedRecordMaxAge, + lastCommitted == null ? -1 : lastCommitted.offset() - 1); + offsetStores.put(recordsTopicPartition, store); + return store; + }); + } } else { uni = Uni.createFrom().item(offsetStore); } diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/OrderedStreamHandler.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/OrderedStreamHandler.java index e26e554581..b762dab455 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/OrderedStreamHandler.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/OrderedStreamHandler.java @@ -154,6 +154,11 @@ public void partitionsRevoked(Collection partitions) { delegate.partitionsRevoked(partitions); } + @Override + public void partitionsSeeked(Collection partitions) { + delegate.partitionsSeeked(partitions); + } + /** * Returns the underlying commit handler. */ diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/ReactiveKafkaConsumer.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/ReactiveKafkaConsumer.java index 116ee6241c..e82a9cc353 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/ReactiveKafkaConsumer.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/ReactiveKafkaConsumer.java @@ -52,7 +52,7 @@ public class ReactiveKafkaConsumer implements io.smallrye.reactive.messagi private final RuntimeKafkaSourceConfiguration configuration; private final Duration pollTimeout; private final String consumerGroup; - private ConsumerRebalanceListener rebalanceListener; + private RebalanceListeners.WrappedConsumerRebalanceListener rebalanceListener; private final AtomicBoolean paused = new AtomicBoolean(); @@ -370,6 +370,7 @@ private void resetPositions(Consumer c, Set partitions) { c.seekToBeginning(Collections.singleton(tp)); } } + notifyCommitHandlerOfSeek(partitions); removeFromQueueRecordsFromTopicPartitions(partitions); c.resume(partitions); } @@ -559,6 +560,7 @@ public Uni> getAssignments() { public Uni seek(TopicPartition partition, long offset) { return runOnPollingThread(c -> { c.seek(partition, offset); + notifyCommitHandlerOfSeek(Collections.singleton(partition)); }); } @@ -567,6 +569,7 @@ public Uni seek(TopicPartition partition, long offset) { public Uni seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata) { return runOnPollingThread(c -> { c.seek(partition, offsetAndMetadata); + notifyCommitHandlerOfSeek(Collections.singleton(partition)); }); } @@ -575,6 +578,7 @@ public Uni seek(TopicPartition partition, OffsetAndMetadata offsetAndMetad public Uni seekToBeginning(Collection partitions) { return runOnPollingThread(c -> { c.seekToBeginning(partitions); + notifyCommitHandlerOfSeek(partitions.isEmpty() ? c.assignment() : partitions); }); } @@ -583,9 +587,16 @@ public Uni seekToBeginning(Collection partitions) { public Uni seekToEnd(Collection partitions) { return runOnPollingThread(c -> { c.seekToEnd(partitions); + notifyCommitHandlerOfSeek(partitions.isEmpty() ? c.assignment() : partitions); }); } + private void notifyCommitHandlerOfSeek(Collection partitions) { + if (rebalanceListener != null) { + rebalanceListener.onPartitionsSeeked(partitions); + } + } + @Override @CheckReturnValue public Uni>> lisTopics() { diff --git a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/RebalanceListeners.java b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/RebalanceListeners.java index 8bb7798678..a9e99f097e 100644 --- a/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/RebalanceListeners.java +++ b/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/RebalanceListeners.java @@ -19,7 +19,7 @@ public class RebalanceListeners { - static ConsumerRebalanceListener createRebalanceListener( + static WrappedConsumerRebalanceListener createRebalanceListener( ReactiveKafkaConsumer reactiveKafkaConsumer, String consumerGroup, KafkaConsumerRebalanceListener listener, @@ -78,6 +78,14 @@ public void onPartitionsAssigned(Collection partitions) { throw e; } } + + public void onPartitionsSeeked(Collection partitions) { + reactiveKafkaConsumer.removeFromQueueRecordsFromTopicPartitions(partitions); + commitHandler.partitionsSeeked(partitions); + if (listener != null) { + listener.onPartitionsSeeked(reactiveKafkaConsumer.unwrap(), partitions); + } + } } public static KafkaConsumerRebalanceListener findMatchingListener( diff --git a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/CommitStrategiesTest.java b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/CommitStrategiesTest.java index 5b13eb84c4..2967d748a9 100644 --- a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/CommitStrategiesTest.java +++ b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/CommitStrategiesTest.java @@ -157,6 +157,182 @@ void testLatestCommitStrategy() { }); } + @Test + void testLatestCommitStrategyWithPartitionsRevoked() { + String group = UUID.randomUUID().toString(); + MapBasedConfig config = commonConfiguration() + .with("commit-strategy", "latest") + .with("lazy-client", true) + .with("client.id", UUID.randomUUID().toString()); + source = createSource(group, config); + injectMockConsumer(source, consumer); + + List> list = new ArrayList<>(); + source.getStream().subscribe().with(list::add); + + TopicPartition tp0 = new TopicPartition(TOPIC, 0); + TopicPartition tp1 = new TopicPartition(TOPIC, 1); + Map beginning = new HashMap<>(); + beginning.put(tp0, 0L); + beginning.put(tp1, 0L); + consumer.updateBeginningOffsets(beginning); + + // Assign both partitions, process records on each + consumer.schedulePollTask(() -> { + consumer.rebalance(Arrays.asList(tp0, tp1)); + for (int i = 0; i < 5; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k", "v0-" + i)); + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, i, "k", "v1-" + i)); + } + }); + + await().until(() -> list.size() == 10); + list.forEach(m -> m.ack().toCompletableFuture().join()); + + await().untilAsserted(() -> { + Map committed = consumer + .committed(new HashSet<>(Arrays.asList(tp0, tp1))); + assertThat(committed.get(tp0).offset()).isEqualTo(5); + assertThat(committed.get(tp1).offset()).isEqualTo(5); + }); + + // Simulate partition revoke/re-assign: clear the commit handler's offsets map + // for tp1, then seek tp1 back to simulate resuming from a lower offset + // (e.g. after a consumer group offset reset or another consumer that committed + // a lower offset while it owned the partition). + source.getCommitHandler().partitionsRevoked(Collections.singletonList(tp1)); + consumer.schedulePollTask(() -> { + consumer.seek(tp1, 0); + for (int i = 0; i < 3; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, i, "k", "v1-replay-" + i)); + } + }); + + await().until(() -> list.size() == 13); + for (int i = 10; i < 13; i++) { + list.get(i).ack().toCompletableFuture().join(); + } + + // With partitionsRevoked clearing the offsets map, the replayed records + // at offsets 0-2 trigger new commits and committed offset becomes 3. + // Without the fix, the stale offset (5) causes record.offset+1 <= 5 + // for all replayed records, so no commits occur and offset stays at 5. + await().untilAsserted(() -> { + Map committed = consumer + .committed(new HashSet<>(Arrays.asList(tp0, tp1))); + assertThat(committed.get(tp0).offset()).isEqualTo(5); + assertThat(committed.get(tp1).offset()).isEqualTo(3); + }); + } + + @Test + void testLatestCommitStrategyWithSeek() { + String group = UUID.randomUUID().toString(); + MapBasedConfig config = commonConfiguration() + .with("commit-strategy", "latest") + .with("lazy-client", true) + .with("client.id", UUID.randomUUID().toString()); + source = createSource(group, config); + injectMockConsumer(source, consumer); + + List> list = new ArrayList<>(); + source.getStream().subscribe().with(list::add); + + TopicPartition tp0 = new TopicPartition(TOPIC, 0); + Map beginning = new HashMap<>(); + beginning.put(tp0, 0L); + consumer.updateBeginningOffsets(beginning); + + consumer.schedulePollTask(() -> { + consumer.rebalance(Collections.singletonList(tp0)); + for (int i = 0; i < 5; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k", "v-" + i)); + } + }); + + await().until(() -> list.size() == 5); + list.forEach(m -> m.ack().toCompletableFuture().join()); + + await().untilAsserted(() -> { + Map committed = consumer.committed(Collections.singleton(tp0)); + assertThat(committed.get(tp0).offset()).isEqualTo(5); + }); + + // Simulate seek via partitionsSeeked (as called by ReactiveKafkaConsumer.seek) + source.getCommitHandler().partitionsSeeked(Collections.singletonList(tp0)); + consumer.schedulePollTask(() -> { + consumer.seek(tp0, 0); + for (int i = 0; i < 3; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k", "v-seek-" + i)); + } + }); + + await().until(() -> list.size() == 8); + for (int i = 5; i < 8; i++) { + list.get(i).ack().toCompletableFuture().join(); + } + + await().untilAsserted(() -> { + Map committed = consumer.committed(Collections.singleton(tp0)); + assertThat(committed.get(tp0).offset()).isEqualTo(3); + }); + } + + @Test + void testThrottledStrategyWithSeek() { + MapBasedConfig config = commonConfiguration() + .with("lazy-client", true) + .with("commit-strategy", "throttled") + .with("auto.commit.interval.ms", 100); + String group = UUID.randomUUID().toString(); + source = createSource(group, config); + injectMockConsumer(source, consumer); + + List> list = new ArrayList<>(); + source.getStream().subscribe().with(list::add); + + TopicPartition tp0 = new TopicPartition(TOPIC, 0); + consumer.updateBeginningOffsets(Collections.singletonMap(tp0, 0L)); + + consumer.schedulePollTask(() -> { + consumer.rebalance(Collections.singletonList(tp0)); + source.getCommitHandler().partitionsAssigned(Collections.singletonList(tp0)); + for (int i = 0; i < 5; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k", "v-" + i)); + } + }); + + await().until(() -> list.size() == 5); + list.forEach(m -> m.ack().toCompletableFuture().join()); + + await().untilAsserted(() -> { + Map committed = consumer.committed(Collections.singleton(tp0)); + assertThat(committed.get(tp0)).isNotNull(); + assertThat(committed.get(tp0).offset()).isEqualTo(5); + }); + + // Simulate seek via partitionsSeeked — must reset the OffsetStore + // so that records below the old lastProcessedOffset are tracked + source.getCommitHandler().partitionsSeeked(Collections.singletonList(tp0)); + consumer.schedulePollTask(() -> { + consumer.seek(tp0, 0); + for (int i = 0; i < 3; i++) { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, i, "k", "v-seek-" + i)); + } + }); + + await().until(() -> list.size() == 8); + for (int i = 5; i < 8; i++) { + list.get(i).ack().toCompletableFuture().join(); + } + + await().untilAsserted(() -> { + Map committed = consumer.committed(Collections.singleton(tp0)); + assertThat(committed.get(tp0)).isNotNull(); + assertThat(committed.get(tp0).offset()).isEqualTo(3); + }); + } + @Test void testThrottledStrategy() { MapBasedConfig config = commonConfiguration() diff --git a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/ConsumerSeekTest.java b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/ConsumerSeekTest.java new file mode 100644 index 0000000000..22ce5256b0 --- /dev/null +++ b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/ConsumerSeekTest.java @@ -0,0 +1,287 @@ +package io.smallrye.reactive.messaging.kafka.commit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.eclipse.microprofile.reactive.messaging.Channel; +import org.eclipse.microprofile.reactive.messaging.Incoming; +import org.junit.jupiter.api.Test; + +import io.smallrye.common.annotation.Identifier; +import io.smallrye.reactive.messaging.PausableChannel; +import io.smallrye.reactive.messaging.annotations.Blocking; +import io.smallrye.reactive.messaging.kafka.KafkaClientService; +import io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener; +import io.smallrye.reactive.messaging.kafka.base.KafkaCompanionTestBase; +import io.smallrye.reactive.messaging.kafka.base.KafkaMapBasedConfig; + +public class ConsumerSeekTest extends KafkaCompanionTestBase { + + private KafkaMapBasedConfig commonConfig(String group, String commitStrategy) { + KafkaMapBasedConfig config = kafkaConfig("mp.messaging.incoming.data"); + config.put("group.id", group); + config.put("topic", topic); + config.put("value.deserializer", IntegerDeserializer.class.getName()); + config.put("enable.auto.commit", "false"); + config.put("auto.offset.reset", "earliest"); + config.put("commit-strategy", commitStrategy); + config.put("pausable", true); + config.put("consumer-rebalance-listener.name", "seek-listener"); + if ("throttled".equals(commitStrategy)) { + config.put("auto.commit.interval.ms", 100); + } + return config; + } + + @Test + public void testSeekToBeginningWithLatestStrategy() { + String group = "test-seek-pausable-latest"; + + addBeans(SeekConsumerBean.class, SeekRebalanceListener.class); + runApplication(commonConfig(group, "latest")); + + SeekConsumerBean bean = get(SeekConsumerBean.class); + SeekRebalanceListener listener = getBeanManager().createInstance() + .select(SeekRebalanceListener.class) + .select(Identifier.Literal.of("seek-listener")) + .get(); + + TopicPartition tp = new TopicPartition(topic, 0); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + await().atMost(30, TimeUnit.SECONDS).until(() -> bean.getCount() >= 10); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + bean.seekToBeginning(Collections.singleton(tp)); + + assertThat(listener.getSeekedPartitions()).contains(tp); + + await().atMost(30, TimeUnit.SECONDS) + .until(() -> bean.getCount() >= 20); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @Test + public void testSeekWithThrottledStrategy() { + String group = "test-seek-pausable-throttled"; + + addBeans(SeekConsumerBean.class, SeekRebalanceListener.class); + runApplication(commonConfig(group, "throttled")); + + SeekConsumerBean bean = get(SeekConsumerBean.class); + SeekRebalanceListener listener = getBeanManager().createInstance() + .select(SeekRebalanceListener.class) + .select(Identifier.Literal.of("seek-listener")) + .get(); + + TopicPartition tp = new TopicPartition(topic, 0); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + await().atMost(30, TimeUnit.SECONDS).until(() -> bean.getCount() >= 10); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + bean.seekToBeginning(Collections.singleton(tp)); + + assertThat(listener.getSeekedPartitions()).contains(tp); + + await().atMost(30, TimeUnit.SECONDS) + .until(() -> bean.getCount() >= 20); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @Test + public void testSeekToBeginningWithOffsetResetInListener() { + String group = "test-seek-with-offset-reset"; + + addBeans(SeekConsumerBean.class, SeekAndResetOffsetListener.class); + + KafkaMapBasedConfig config = kafkaConfig("mp.messaging.incoming.data"); + config.put("group.id", group); + config.put("topic", topic); + config.put("value.deserializer", IntegerDeserializer.class.getName()); + config.put("enable.auto.commit", "false"); + config.put("auto.offset.reset", "earliest"); + config.put("commit-strategy", "latest"); + config.put("pausable", true); + config.put("consumer-rebalance-listener.name", "seek-and-reset-listener"); + + runApplication(config); + + SeekConsumerBean bean = get(SeekConsumerBean.class); + + TopicPartition tp = new TopicPartition(topic, 0); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + await().atMost(30, TimeUnit.SECONDS).until(() -> bean.getCount() >= 10); + + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + int countBeforeSeek = bean.getCount(); + + // The listener resets the committed offset to the seeked position + // in onPartitionsSeeked, so replayed records MUST be re-committed. + // Without partitionsSeeked clearing the commit handler state, + // the stale offset (10) would block all commits for records 0-9, + // and the committed offset would stay at 0. + bean.seekToBeginning(Collections.singleton(tp)); + + await().atMost(30, TimeUnit.SECONDS) + .until(() -> bean.getCount() >= countBeforeSeek + 10); + + // Committed offset must advance back to 10 + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + } + + @ApplicationScoped + public static class SeekConsumerBean { + + @Inject + @Channel("data") + PausableChannel pausable; + + @Inject + KafkaClientService clientService; + + private final List received = new CopyOnWriteArrayList<>(); + private final AtomicInteger wip = new AtomicInteger(); + + @Incoming("data") + @Blocking + public void consume(int payload) { + wip.incrementAndGet(); + try { + received.add(payload); + } finally { + wip.decrementAndGet(); + } + } + + public void seekToBeginning(Collection partitions) { + pausable.pause(); + await().until(() -> wip.get() == 0); + clientService. getConsumer("data") + .seekToBeginning(partitions).await().indefinitely(); + pausable.clearBuffer(); + pausable.resume(); + } + + public int getCount() { + return received.size(); + } + + public List getReceived() { + return received; + } + } + + @ApplicationScoped + @Identifier("seek-listener") + public static class SeekRebalanceListener implements KafkaConsumerRebalanceListener { + + private final Set seekedPartitions = Collections.newSetFromMap(new ConcurrentHashMap<>()); + + @Override + public void onPartitionsSeeked(Consumer consumer, Collection partitions) { + seekedPartitions.addAll(partitions); + } + + public Set getSeekedPartitions() { + return seekedPartitions; + } + } + + @ApplicationScoped + @Identifier("seek-and-reset-listener") + public static class SeekAndResetOffsetListener implements KafkaConsumerRebalanceListener { + + @Override + public void onPartitionsSeeked(Consumer consumer, Collection partitions) { + Map offsets = new HashMap<>(); + for (TopicPartition tp : partitions) { + offsets.put(tp, new OffsetAndMetadata(consumer.position(tp))); + } + consumer.commitSync(offsets); + } + } +} diff --git a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandlerTest.java b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandlerTest.java index 4e5bdc742f..76cb58157f 100644 --- a/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandlerTest.java +++ b/smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandlerTest.java @@ -9,6 +9,7 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; @@ -16,24 +17,36 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; +import jakarta.enterprise.context.ApplicationScoped; + +import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.eclipse.microprofile.reactive.messaging.Incoming; import org.eclipse.microprofile.reactive.messaging.Message; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import io.smallrye.common.annotation.Identifier; +import io.smallrye.mutiny.Uni; import io.smallrye.reactive.messaging.health.HealthReport; import io.smallrye.reactive.messaging.kafka.IncomingKafkaRecord; +import io.smallrye.reactive.messaging.kafka.KafkaClientService; +import io.smallrye.reactive.messaging.kafka.KafkaConsumer; +import io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener; import io.smallrye.reactive.messaging.kafka.KafkaRecord; import io.smallrye.reactive.messaging.kafka.base.KafkaCompanionTestBase; +import io.smallrye.reactive.messaging.kafka.base.KafkaMapBasedConfig; import io.smallrye.reactive.messaging.kafka.impl.KafkaSource; import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig; @@ -352,4 +365,391 @@ void testSourceWithThrottledAndRebalanceWithPartitionsConfig() { await().atMost(2, TimeUnit.MINUTES).until(() -> messages1.size() + messages2.size() >= 20000); } + + @Test + public void testSourceWithLatestAndRebalance() { + String group = "test-source-with-latest-commit-and-rebalance"; + companion.topics().createAndWait(topic, 2); + + MapBasedConfig config1 = newCommonConfigForSource() + .with("client.id", UUID.randomUUID().toString()) + .with("group.id", group) + .with("value.deserializer", IntegerDeserializer.class.getName()) + .with("commit-strategy", "latest"); + + // Source 2 uses "ignore" — receives records but never commits offsets + MapBasedConfig config2 = newCommonConfigForSource() + .with("client.id", UUID.randomUUID().toString()) + .with("group.id", group) + .with("value.deserializer", IntegerDeserializer.class.getName()) + .with("commit-strategy", "ignore"); + + source = createSource(group, config1); + + List> messages1 = new CopyOnWriteArrayList<>(); + source.getStream() + .onItem().call(m -> Uni.createFrom().completionStage(m.ack())) + .subscribe().with(messages1::add); + + // Source 1 gets both partitions + await().until(() -> source.getConsumer().getAssignments().await().indefinitely().size() == 2); + + TopicPartition tp0 = new TopicPartition(topic, 0); + TopicPartition tp1 = new TopicPartition(topic, 1); + + // Batch 1: 100 records, all processed and committed by source 1 + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, Integer.toString(i % 2), i), 100); + + await().atMost(2, TimeUnit.MINUTES).until(() -> messages1.size() >= 100); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + Map result = companion.consumerGroups().offsets( + group, Arrays.asList(tp0, tp1)); + assertNotNull(result.get(tp0)); + assertNotNull(result.get(tp1)); + assertEquals(100, result.get(tp0).offset() + result.get(tp1).offset()); + }); + + // Source 2 joins with "ignore" strategy — triggers rebalance, partitions split + KafkaSource source2 = createSource(group, config2); + List> messages2 = new CopyOnWriteArrayList<>(); + source2.getStream().subscribe().with(messages2::add); + + await().until(() -> source2.getConsumer().getAssignments().await().indefinitely().size() == 1 + && source.getConsumer().getAssignments().await().indefinitely().size() == 1); + + // Batch 2: 100 records while source 2 is active + // Source 2 receives records on its partition but never commits + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, Integer.toString(i % 2), i), 100); + + await().atMost(2, TimeUnit.MINUTES).until(() -> messages2.size() >= 10); + + // Close source 2 — triggers rebalance, source 1 gets both partitions back. + // Source 2 never committed, so for its partition the committed offset is still + // from batch 1. Source 1 replays batch-2 records on that partition. + source2.closeQuietly(); + + await().until(() -> source.getConsumer().getAssignments().await().indefinitely().size() == 2); + + // Committed offsets must cover all 200 produced records. + // partitionsRevoked clears the stale offsets map so that replayed records + // are correctly committed after the partition is reassigned. + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + Map result = companion.consumerGroups().offsets( + group, Arrays.asList(tp0, tp1)); + assertNotNull(result.get(tp0)); + assertNotNull(result.get(tp1)); + assertEquals(200, result.get(tp0).offset() + result.get(tp1).offset()); + }); + } + + @Test + public void testSourceWithLatestAndSeekToBeginning() { + String group = "test-source-with-latest-and-seek"; + + MapBasedConfig config = newCommonConfigForSource() + .with("client.id", UUID.randomUUID().toString()) + .with("group.id", group) + .with("value.deserializer", IntegerDeserializer.class.getName()) + .with("commit-strategy", "latest"); + + source = createSource(group, config); + + List> messages = new CopyOnWriteArrayList<>(); + source.getStream() + .onItem().call(m -> Uni.createFrom().completionStage(m.ack())) + .subscribe().with(messages::add); + + TopicPartition tp = new TopicPartition(topic, 0); + + // Produce and consume 10 records + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + int countBeforeSeek = messages.size(); + + // Seek to beginning through the consumer API — full path: + // ReactiveKafkaConsumer → WrappedConsumerRebalanceListener.onPartitionsSeeked + // → buffer clear + commitHandler.partitionsSeeked + source.getConsumer().seekToBeginning(Collections.singleton(tp)) + .await().indefinitely(); + + // Records should be re-delivered from offset 0 + await().atMost(2, TimeUnit.MINUTES) + .until(() -> messages.size() >= countBeforeSeek + 10); + + // After seek + replay, committed offset must still be correct + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + // Produce 10 more records (offsets 10-19) + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @Test + public void testSourceWithThrottledAndSeek() { + String group = "test-source-with-throttled-and-seek"; + + MapBasedConfig config = newCommonConfigForSource() + .with("client.id", UUID.randomUUID().toString()) + .with("group.id", group) + .with("value.deserializer", IntegerDeserializer.class.getName()) + .with("commit-strategy", "throttled") + .with(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); + + source = createSource(group, config); + + List> messages = new CopyOnWriteArrayList<>(); + source.getStream() + .onItem().call(m -> Uni.createFrom().completionStage(m.ack())) + .subscribe().with(messages::add); + + TopicPartition tp = new TopicPartition(topic, 0); + + // Produce and consume 10 records + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + int countBeforeSeek = messages.size(); + + // Seek to offset 0 through the consumer API + source.getConsumer().seek(tp, 0).await().indefinitely(); + + // Records re-delivered from offset 0 + await().atMost(2, TimeUnit.MINUTES) + .until(() -> messages.size() >= countBeforeSeek + 10); + + // After seek + replay, committed offset must still be correct + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + // Produce 10 more records + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @Test + public void testSeekToBeginningWithApplicationScopedBean() { + String group = "test-app-seek-to-beginning"; + + addBeans(SeekConsumerBean.class, SeekTrackingRebalanceListener.class); + + KafkaMapBasedConfig config = kafkaConfig("mp.messaging.incoming.data"); + config.put("group.id", group); + config.put("topic", topic); + config.put("value.deserializer", IntegerDeserializer.class.getName()); + config.put("auto.offset.reset", "earliest"); + config.put("commit-strategy", "latest"); + config.put("consumer-rebalance-listener.name", "seek-tracking-listener"); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + runApplication(config); + + SeekConsumerBean bean = get(SeekConsumerBean.class); + SeekTrackingRebalanceListener listener = getBeanManager().createInstance() + .select(SeekTrackingRebalanceListener.class) + .select(Identifier.Literal.of("seek-tracking-listener")) + .get(); + + await().atMost(2, TimeUnit.MINUTES).until(() -> bean.getCount() >= 10); + + TopicPartition tp = new TopicPartition(topic, 0); + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + // Seek to beginning via KafkaClientService — exercises the full CDI path + KafkaClientService clientService = get(KafkaClientService.class); + KafkaConsumer consumer = clientService.getConsumer("data"); + assertNotNull(consumer); + + int countBeforeSeek = bean.getCount(); + + consumer.seekToBeginning(Collections.singleton(tp)).await().indefinitely(); + + // Verify onPartitionsSeeked was called on the user's rebalance listener + await().atMost(10, TimeUnit.SECONDS) + .until(() -> listener.getSeekedPartitions().contains(tp)); + + // Records should be re-delivered from offset 0 + await().atMost(2, TimeUnit.MINUTES) + .until(() -> bean.getCount() >= countBeforeSeek + 10); + + // After seek + replay, committed offset must still be correct + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + // Produce more records and verify committed offset covers everything + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @Test + public void testSeekWithApplicationScopedBeanAndThrottledStrategy() { + String group = "test-app-seek-throttled"; + + addBeans(SeekConsumerBean.class, SeekTrackingRebalanceListener.class); + + KafkaMapBasedConfig config = kafkaConfig("mp.messaging.incoming.data"); + config.put("group.id", group); + config.put("topic", topic); + config.put("value.deserializer", IntegerDeserializer.class.getName()); + config.put("auto.offset.reset", "earliest"); + config.put("commit-strategy", "throttled"); + config.put("auto.commit.interval.ms", 100); + config.put("consumer-rebalance-listener.name", "seek-tracking-listener"); + + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, i), 10); + + runApplication(config); + + SeekConsumerBean bean = get(SeekConsumerBean.class); + SeekTrackingRebalanceListener listener = getBeanManager().createInstance() + .select(SeekTrackingRebalanceListener.class) + .select(Identifier.Literal.of("seek-tracking-listener")) + .get(); + + await().atMost(2, TimeUnit.MINUTES).until(() -> bean.getCount() >= 10); + + TopicPartition tp = new TopicPartition(topic, 0); + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + KafkaClientService clientService = get(KafkaClientService.class); + KafkaConsumer consumer = clientService.getConsumer("data"); + assertNotNull(consumer); + + int countBeforeSeek = bean.getCount(); + + // Seek to offset 0 via consumer API — full path through throttled handler + consumer.seek(tp, 0).await().indefinitely(); + + // Verify onPartitionsSeeked was called + await().atMost(10, TimeUnit.SECONDS) + .until(() -> listener.getSeekedPartitions().contains(tp)); + + // Records re-delivered from offset 0 + await().atMost(2, TimeUnit.MINUTES) + .until(() -> bean.getCount() >= countBeforeSeek + 10); + + // After seek + replay, committed offset must still be correct + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(10L, offset.offset()); + }); + + // Produce more records and verify committed offset advances + companion.produceIntegers() + .usingGenerator(i -> new ProducerRecord<>(topic, 10 + i), 10); + + await().atMost(2, TimeUnit.MINUTES) + .untilAsserted(() -> { + OffsetAndMetadata offset = companion.consumerGroups().offsets(group, tp); + assertNotNull(offset); + assertEquals(20L, offset.offset()); + }); + } + + @ApplicationScoped + public static class SeekConsumerBean { + + private final List received = new CopyOnWriteArrayList<>(); + + @Incoming("data") + public void consume(int payload) { + received.add(payload); + } + + public int getCount() { + return received.size(); + } + + public List getReceived() { + return received; + } + } + + @ApplicationScoped + @Identifier("seek-tracking-listener") + public static class SeekTrackingRebalanceListener implements KafkaConsumerRebalanceListener { + + private final Set seekedPartitions = Collections.newSetFromMap(new ConcurrentHashMap<>()); + + @Override + public void onPartitionsSeeked(Consumer consumer, Collection partitions) { + seekedPartitions.addAll(partitions); + } + + public Set getSeekedPartitions() { + return seekedPartitions; + } + } } diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/PausableChannelDecorator.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/PausableChannelDecorator.java index fcf5214109..0c1070bf57 100644 --- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/PausableChannelDecorator.java +++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/PausableChannelDecorator.java @@ -91,6 +91,11 @@ public void pause() { public void resume() { pauser.resume(); } + + @Override + public void clearBuffer() { + pauser.clearBuffer(); + } } }