draft application thread to tell background thread which call the event

apache · Nov 23, 2024 · bb67c22 · bb67c22
1 parent 3a79ae4
commit bb67c22
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 5 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
@@ -904,7 +904,7 @@ public long position(TopicPartition partition, Duration timeout) {
                 if (position != null)
                     return position.offset;
 
-                updateFetchPositions(timer);
+                updateFetchPositions(timer, false);
                 timer.update();
                 wakeupTrigger.maybeTriggerWakeup();
             } while (timer.notExpired());
@@ -1683,10 +1683,10 @@ private Fetch<K, V> collectFetch() {
      * @throws NoOffsetForPartitionException If no offset is stored for a given partition and no offset reset policy is
      *                                       defined
      */
-    private boolean updateFetchPositions(final Timer timer) {
+    private boolean updateFetchPositions(final Timer timer, final boolean isCompletedBuFuture) {
         cachedSubscriptionHasAllFetchPositions = false;
         try {
-            CheckAndUpdatePositionsEvent checkAndUpdatePositionsEvent = new CheckAndUpdatePositionsEvent(calculateDeadlineMs(timer));
+            CheckAndUpdatePositionsEvent checkAndUpdatePositionsEvent = new CheckAndUpdatePositionsEvent(calculateDeadlineMs(timer), isCompletedBuFuture);
             wakeupTrigger.setActiveTask(checkAndUpdatePositionsEvent.future());
             cachedSubscriptionHasAllFetchPositions = applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
         } catch (TimeoutException e) {
@@ -1775,7 +1775,7 @@ public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
         }
         processBackgroundEvents();
 
-        return updateFetchPositions(timer);
+        return updateFetchPositions(timer, true);
     }
 
     @Override

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
@@ -20,6 +20,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
 import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
 import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent;
+import org.apache.kafka.clients.consumer.internals.events.CompletableApplicationEvent;
 import org.apache.kafka.clients.consumer.internals.events.CompletableEvent;
 import org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper;
 import org.apache.kafka.common.internals.IdempotentCloser;
@@ -157,7 +158,9 @@ void runOnce() {
 
         if (networkClientDelegate.metadataError().isPresent()) {
             Throwable metadataError = networkClientDelegate.metadataError().get();
-            completableEvents.forEach(event -> event.future().completeExceptionally(metadataError));
+            completableEvents.stream()
+                    .filter(event -> !(event instanceof CompletableApplicationEvent && ((CompletableApplicationEvent<?>) event).isCompletedByFuture()))
+                    .forEach(event -> event.future().completeExceptionally(metadataError));
             networkClientDelegate.clearMetadataError();
         }
     }

diff --git a/...java/org/apache/kafka/clients/consumer/internals/events/CheckAndUpdatePositionsEvent.java b/...java/org/apache/kafka/clients/consumer/internals/events/CheckAndUpdatePositionsEvent.java
@@ -32,4 +32,8 @@ public class CheckAndUpdatePositionsEvent extends CompletableApplicationEvent<Bo
     public CheckAndUpdatePositionsEvent(long deadlineMs) {
         super(Type.CHECK_AND_UPDATE_POSITIONS, deadlineMs);
     }
+
+    public CheckAndUpdatePositionsEvent(long deadlineMs, boolean isCompletedByFuture) {
+        super(Type.CHECK_AND_UPDATE_POSITIONS, deadlineMs, isCompletedByFuture);
+    }
 }
diff --git a/.../java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java b/.../java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java
@@ -28,6 +28,7 @@ public abstract class CompletableApplicationEvent<T> extends ApplicationEvent im
 
     private final CompletableFuture<T> future;
     private final long deadlineMs;
+    private boolean isCompletedByFuture = false;
 
     /**
      * <em>Note</em>: the {@code deadlineMs} is the future time of expiration, <em>not</em> a timeout.
@@ -38,6 +39,13 @@ protected CompletableApplicationEvent(final Type type, final long deadlineMs) {
         this.deadlineMs = deadlineMs;
     }
 
+    protected CompletableApplicationEvent(final Type type, final long deadlineMs, final boolean isCompletedByFuture) {
+        super(type);
+        this.future = new CompletableFuture<>();
+        this.deadlineMs = deadlineMs;
+        this.isCompletedByFuture = isCompletedByFuture;
+    }
+
     @Override
     public CompletableFuture<T> future() {
         return future;
@@ -52,4 +60,8 @@ public long deadlineMs() {
     protected String toStringBase() {
         return super.toStringBase() + ", future=" + future + ", deadlineMs=" + deadlineMs;
     }
+
+    public boolean isCompletedByFuture() {
+        return isCompletedByFuture;
+    }
 }