Kafka handle position and commit offset on seek operations

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() {
+    }
 }

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/KafkaConsumerRebalanceListener.java

@@ -142,4 +142,23 @@ default void onPartitionsLost(Consumer<?, ?> consumer, Collection<org.apache.kaf
         onPartitionsRevoked(consumer, partitions);
     }
 
+    /**
+     * Called after the consumer position is explicitly changed via seek operations
+     * ({@link Consumer#seek}, {@link Consumer#seekToBeginning}, {@link Consumer#seekToEnd}).
+     * <p>
+     * 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.
+     * <p>
+     * This allows listeners to reset any per-partition state that depends on the consumer's position.
+     * <p>
+     * <strong>IMPORTANT</strong>: 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<org.apache.kafka.common.TopicPartition> partitions) {
+
+    }
+
 }

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/commit/KafkaCommitHandler.java

@@ -94,6 +94,16 @@ default void partitionsRevoked(Collection<TopicPartition> 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<TopicPartition> partitions) {
+        // Do nothing by default.
+    }
+
     /**
      * Handle message acknowledgment
      *

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<TopicPartition> partitions) {
+        runOnContextAndAwait(() -> {
+            for (TopicPartition partition : partitions) {
+                offsets.remove(partition);
+            }
+            return null;
+        });
+    }
+
+    @Override
+    public void partitionsSeeked(Collection<TopicPartition> partitions) {
+        partitionsRevoked(partitions);
+    }
+
     @Override
     public <K, V> Uni<Void> handle(IncomingKafkaRecord<K, V> record) {
         runOnContext(() -> {

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<TopicPartition, OffsetStore> offsetStores = new HashMap<>();
+    private final Set<TopicPartition> seekedPartitions = new HashSet<>();
 
     private final String groupId;
     private final KafkaConsumer<?, ?> consumer;
@@ -146,6 +147,7 @@ public void partitionsRevoked(Collection<TopicPartition> partitions) {
             Map<TopicPartition, OffsetAndMetadata> 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<TopicPartition> partitions) {
         }
     }
 
+    @Override
+    public void partitionsSeeked(Collection<TopicPartition> 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 <K, V> Uni<IncomingKafkaRecord<K, V>> received(IncomingKafkaRecord<K, V>
         OffsetStore offsetStore = offsetStores.get(recordsTopicPartition);
         Uni<OffsetStore> 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);
         }

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/OrderedStreamHandler.java

@@ -154,6 +154,11 @@ public void partitionsRevoked(Collection<TopicPartition> partitions) {
         delegate.partitionsRevoked(partitions);
     }
 
+    @Override
+    public void partitionsSeeked(Collection<TopicPartition> partitions) {
+        delegate.partitionsSeeked(partitions);
+    }
+
     /**
      * Returns the underlying commit handler.
      */

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/ReactiveKafkaConsumer.java

@@ -52,7 +52,7 @@ public class ReactiveKafkaConsumer<K, V> 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<K, V> c, Set<TopicPartition> partitions) {
                 c.seekToBeginning(Collections.singleton(tp));
             }
         }
+        notifyCommitHandlerOfSeek(partitions);
         removeFromQueueRecordsFromTopicPartitions(partitions);
         c.resume(partitions);
     }
@@ -559,6 +560,7 @@ public Uni<Set<TopicPartition>> getAssignments() {
     public Uni<Void> seek(TopicPartition partition, long offset) {
         return runOnPollingThread(c -> {
             c.seek(partition, offset);
+            notifyCommitHandlerOfSeek(Collections.singleton(partition));
         });
     }
 
@@ -567,6 +569,7 @@ public Uni<Void> seek(TopicPartition partition, long offset) {
     public Uni<Void> seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata) {
         return runOnPollingThread(c -> {
             c.seek(partition, offsetAndMetadata);
+            notifyCommitHandlerOfSeek(Collections.singleton(partition));
         });
     }
 
@@ -575,6 +578,7 @@ public Uni<Void> seek(TopicPartition partition, OffsetAndMetadata offsetAndMetad
     public Uni<Void> seekToBeginning(Collection<TopicPartition> partitions) {
         return runOnPollingThread(c -> {
             c.seekToBeginning(partitions);
+            notifyCommitHandlerOfSeek(partitions.isEmpty() ? c.assignment() : partitions);
         });
     }
 
@@ -583,9 +587,16 @@ public Uni<Void> seekToBeginning(Collection<TopicPartition> partitions) {
     public Uni<Void> seekToEnd(Collection<TopicPartition> partitions) {
         return runOnPollingThread(c -> {
             c.seekToEnd(partitions);
+            notifyCommitHandlerOfSeek(partitions.isEmpty() ? c.assignment() : partitions);
         });
     }
 
+    private void notifyCommitHandlerOfSeek(Collection<TopicPartition> partitions) {
+        if (rebalanceListener != null) {
+            rebalanceListener.onPartitionsSeeked(partitions);
+        }
+    }
+
     @Override
     @CheckReturnValue
     public Uni<Map<String, List<PartitionInfo>>> lisTopics() {

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<TopicPartition> partitions) {
                 throw e;
             }
         }
+
+        public void onPartitionsSeeked(Collection<TopicPartition> partitions) {
+            reactiveKafkaConsumer.removeFromQueueRecordsFromTopicPartitions(partitions);
+            commitHandler.partitionsSeeked(partitions);
+            if (listener != null) {
+                listener.onPartitionsSeeked(reactiveKafkaConsumer.unwrap(), partitions);
+            }
+        }
     }
 
     public static KafkaConsumerRebalanceListener findMatchingListener(