Remove offsetManager in ResetOffsetEvent

apache · Sep 29, 2024 · f8379a0 · f8379a0
1 parent 8e10de1
commit f8379a0
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 8 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -833,7 +833,7 @@ public void seekToBeginning(Collection<TopicPartition> partitions) {
             Timer timer = time.timer(defaultApiTimeoutMs);
             ResetOffsetEvent resetOffsetEvent = new ResetOffsetEvent(partitions, OffsetResetStrategy.EARLIEST,
                     calculateDeadlineMs(timer));
-            cachedSubscriptionHasAllFetchPositions = applicationEventHandler.addAndGet(resetOffsetEvent);
+            applicationEventHandler.addAndGet(resetOffsetEvent);
         } finally {
             release();
         }
@@ -849,7 +849,7 @@ public void seekToEnd(Collection<TopicPartition> partitions) {
             Timer timer = time.timer(defaultApiTimeoutMs);
             ResetOffsetEvent resetOffsetEvent = new ResetOffsetEvent(partitions, OffsetResetStrategy.LATEST,
                     calculateDeadlineMs(timer));
-            cachedSubscriptionHasAllFetchPositions = applicationEventHandler.addAndGet(resetOffsetEvent);
+            applicationEventHandler.addAndGet(resetOffsetEvent);
         } finally {
             release();
         }

diff --git a/...in/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/...in/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
@@ -261,11 +261,14 @@ private void process(final UnsubscribeEvent event) {
     }
 
     private void process(final ResetOffsetEvent event) {
-        Collection<TopicPartition> parts = event.topicPartitions().isEmpty() ?
-                subscriptions.assignedPartitions() : event.topicPartitions();
-        subscriptions.requestOffsetReset(parts, event.offsetResetStrategy());
-        CompletableFuture<Boolean> future = requestManagers.offsetsRequestManager.updateFetchPositions(event.deadlineMs());
-        future.whenComplete(complete(event.future()));
+        try {
+            Collection<TopicPartition> parts = event.topicPartitions().isEmpty() ?
+                    subscriptions.assignedPartitions() : event.topicPartitions();
+            subscriptions.requestOffsetReset(parts, event.offsetResetStrategy());
+            event.future().complete(null);
+        } catch (Exception e) {
+            event.future().completeExceptionally(e);
+        }
     }
 
     /**

diff --git a/...ts/src/main/java/org/apache/kafka/clients/consumer/internals/events/ResetOffsetEvent.java b/...ts/src/main/java/org/apache/kafka/clients/consumer/internals/events/ResetOffsetEvent.java
@@ -29,7 +29,7 @@
  * Event to perform {@link AsyncKafkaConsumer#seekToBeginning(Collection)} and {@link AsyncKafkaConsumer#seekToEnd(Collection)}
  * in the background thread. This can avoid race conditions when subscription state is updated.
  */
-public class ResetOffsetEvent extends CompletableApplicationEvent<Boolean> {
+public class ResetOffsetEvent extends CompletableApplicationEvent<Void> {
 
     private final Collection<TopicPartition> topicPartitions;