KAFKA-15974: Enforce that event processing respects user-provided timeout

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

@@ -20,6 +20,8 @@
 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.CompletableEventReaper;
 import org.apache.kafka.common.internals.IdempotentCloser;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.utils.KafkaThread;
@@ -31,10 +33,13 @@
 import java.io.Closeable;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS;
 import static org.apache.kafka.common.utils.Utils.closeQuietly;
@@ -50,6 +55,8 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
     private static final String BACKGROUND_THREAD_NAME = "consumer_background_thread";
     private final Time time;
     private final Logger log;
+    private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+    private final CompletableEventReaper<CompletableApplicationEvent<?>> applicationEventReaper;
     private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
     private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
     private final Supplier<RequestManagers> requestManagersSupplier;
@@ -63,12 +70,15 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
 
     public ConsumerNetworkThread(LogContext logContext,
                                  Time time,
+                                 BlockingQueue<ApplicationEvent> applicationEventQueue,
                                  Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
                                  Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
                                  Supplier<RequestManagers> requestManagersSupplier) {
         super(BACKGROUND_THREAD_NAME, true);
         this.time = time;
         this.log = logContext.logger(getClass());
+        this.applicationEventQueue = applicationEventQueue;
+        this.applicationEventReaper = new CompletableEventReaper<>(logContext);
         this.applicationEventProcessorSupplier = applicationEventProcessorSupplier;
         this.networkClientDelegateSupplier = networkClientDelegateSupplier;
         this.requestManagersSupplier = requestManagersSupplier;
@@ -125,10 +135,7 @@ void initializeResources() {
      * </ol>
      */
     void runOnce() {
-        // Process the events—if any—that were produced by the application thread. It is possible that when processing
-        // an event generates an error. In such cases, the processor will log an exception, but we do not want those
-        // errors to be propagated to the caller.
-        applicationEventProcessor.process();
+        processApplicationEvents();
 
         final long currentTimeMs = time.milliseconds();
         final long pollWaitTimeMs = requestManagers.entries().stream()
@@ -144,6 +151,36 @@ void runOnce() {
                 .map(Optional::get)
                 .map(rm -> rm.maximumTimeToWait(currentTimeMs))
                 .reduce(Long.MAX_VALUE, Math::min);
+
+        reapExpiredApplicationEvents(currentTimeMs);
+    }
+
+    /**
+     * Process the events—if any—that were produced by the application thread.
+     */
+    private void processApplicationEvents() {
+        LinkedList<ApplicationEvent> events = new LinkedList<>();
+        applicationEventQueue.drainTo(events);
+
+        for (ApplicationEvent event : events) {
+            try {
+                if (event instanceof CompletableApplicationEvent)
+                    applicationEventReaper.add((CompletableApplicationEvent<?>) event);
+
+                applicationEventProcessor.process(event);
+            } catch (Throwable t) {
+                log.warn("Error processing event {}", t.getMessage(), t);
+            }
+        }
+    }
+
+    /**
+     * "Complete" any events that have expired. This cleanup step should only be called after the network I/O
+     * thread has made at least one call to {@link NetworkClientDelegate#poll(long, long) poll} so that each event
+     * is given least one attempt to satisfy any network requests <em>before</em> checking if a timeout has expired.
+     */
+    private void reapExpiredApplicationEvents(long currentTimeMs) {
+        applicationEventReaper.reapExpiredAndCompleted(currentTimeMs);
     }
 
     /**
@@ -273,9 +310,18 @@ void cleanup() {
             log.error("Unexpected error during shutdown.  Proceed with closing.", e);
         } finally {
             sendUnsentRequests(timer);
+
+            LinkedList<ApplicationEvent> allEvents = new LinkedList<>();
+            applicationEventQueue.drainTo(allEvents);
+            List<CompletableApplicationEvent<?>> completableEvents = allEvents
+                    .stream()
+                    .filter(e -> e instanceof CompletableApplicationEvent<?>)
+                    .map(e -> (CompletableApplicationEvent<?>) e)
+                    .collect(Collectors.toList());
+            applicationEventReaper.reapIncomplete(completableEvents);
+
             closeQuietly(requestManagers, "request managers");
             closeQuietly(networkClientDelegate, "network client delegate");
-            closeQuietly(applicationEventProcessor, "application event processor");
             log.debug("Closed the consumer network thread");
         }
     }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java

@@ -1338,6 +1338,7 @@ private CompletableFuture<Void> enqueueConsumerRebalanceListenerCallback(Consume
                                                                              Set<TopicPartition> partitions) {
         SortedSet<TopicPartition> sortedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
         sortedPartitions.addAll(partitions);
+
         CompletableBackgroundEvent<Void> event = new ConsumerRebalanceListenerCallbackNeededEvent(methodName, sortedPartitions);
         backgroundEventHandler.add(event);
         log.debug("The event to trigger the {} method execution was enqueued successfully", methodName.fullyQualifiedMethodName());

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/AbstractTopicMetadataEvent.java

@@ -17,14 +17,13 @@
 package org.apache.kafka.clients.consumer.internals.events;
 
 import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.utils.Timer;
 
 import java.util.List;
 import java.util.Map;
 
 public abstract class AbstractTopicMetadataEvent extends CompletableApplicationEvent<Map<String, List<PartitionInfo>>> {
 
-    protected AbstractTopicMetadataEvent(final Type type, final Timer timer) {
-        super(type, timer);
+    protected AbstractTopicMetadataEvent(final Type type, final long deadlineMs) {
+        super(type, deadlineMs);
     }
 }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/AllTopicsMetadataEvent.java

@@ -16,11 +16,9 @@
  */
 package org.apache.kafka.clients.consumer.internals.events;
 
-import org.apache.kafka.common.utils.Timer;
-
 public class AllTopicsMetadataEvent extends AbstractTopicMetadataEvent {
 
-    public AllTopicsMetadataEvent(final Timer timer) {
-        super(Type.ALL_TOPICS_METADATA, timer);
+    public AllTopicsMetadataEvent(final long deadlineMs) {
+        super(Type.ALL_TOPICS_METADATA, deadlineMs);
     }
 }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java

@@ -23,16 +23,13 @@
 import org.apache.kafka.common.internals.IdempotentCloser;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Timer;
 import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
 
 import java.io.Closeable;
 import java.time.Duration;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 /**
@@ -56,6 +53,7 @@ public ApplicationEventHandler(final LogContext logContext,
         this.applicationEventQueue = applicationEventQueue;
         this.networkThread = new ConsumerNetworkThread(logContext,
                 time,
+                applicationEventQueue,
                 applicationEventProcessorSupplier,
                 networkClientDelegateSupplier,
                 requestManagersSupplier);
@@ -98,19 +96,14 @@ public long maximumTimeToWait() {
      * Add a {@link CompletableApplicationEvent} to the handler. The method blocks waiting for the result, and will
      * return the result value upon successful completion; otherwise throws an error.
      *
-     * <p/>
-     *
-     * See {@link ConsumerUtils#getResult(Future, Timer)} and {@link Future#get(long, TimeUnit)} for more details.
-     *
      * @param event A {@link CompletableApplicationEvent} created by the polling thread
      * @return      Value that is the result of the event
      * @param <T>   Type of return value of the event
      */
-    public <T> T addAndGet(final CompletableApplicationEvent<T> event, final Timer timer) {
+    public <T> T addAndGet(final CompletableApplicationEvent<T> event) {
         Objects.requireNonNull(event, "CompletableApplicationEvent provided to addAndGet must be non-null");
-        Objects.requireNonNull(timer, "Timer provided to addAndGet must be non-null");
         add(event);
-        return ConsumerUtils.getResult(event.future(), timer);
+        return ConsumerUtils.getResult(event.future());
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -33,7 +33,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
@@ -42,31 +41,20 @@
  * An {@link EventProcessor} that is created and executes in the {@link ConsumerNetworkThread network thread}
  * which processes {@link ApplicationEvent application events} generated by the application thread.
  */
-public class ApplicationEventProcessor extends EventProcessor<ApplicationEvent> {
+public class ApplicationEventProcessor implements EventProcessor<ApplicationEvent> {
 
     private final Logger log;
     private final ConsumerMetadata metadata;
     private final RequestManagers requestManagers;
 
     public ApplicationEventProcessor(final LogContext logContext,
-                                     final BlockingQueue<ApplicationEvent> applicationEventQueue,
                                      final RequestManagers requestManagers,
                                      final ConsumerMetadata metadata) {
-        super(logContext, applicationEventQueue);
         this.log = logContext.logger(ApplicationEventProcessor.class);
         this.requestManagers = requestManagers;
         this.metadata = metadata;
     }
 
-    /**
-     * Process the events—if any—that were produced by the application thread. It is possible that when processing
-     * an event generates an error. In such cases, the processor will log an exception, but we do not want those
-     * errors to be propagated to the caller.
-     */
-    public boolean process() {
-        return process((event, error) -> error.ifPresent(e -> log.warn("Error processing event {}", e.getMessage(), e)));
-    }
-
     @SuppressWarnings({"CyclomaticComplexity"})
     @Override
     public void process(ApplicationEvent event) {
@@ -273,7 +261,7 @@ private void process(final ConsumerRebalanceListenerCallbackCompletedEvent event
         manager.consumerRebalanceListenerCallbackCompleted(event);
     }
 
-    private void process(final CommitOnCloseEvent event) {
+    private void process(@SuppressWarnings("unused") final CommitOnCloseEvent event) {
         if (!requestManagers.commitRequestManager.isPresent())
             return;
         log.debug("Signal CommitRequestManager closing");
@@ -309,15 +297,13 @@ private void process(final LeaveOnCloseEvent event) {
      */
     public static Supplier<ApplicationEventProcessor> supplier(final LogContext logContext,
                                                                final ConsumerMetadata metadata,
-                                                               final BlockingQueue<ApplicationEvent> applicationEventQueue,
                                                                final Supplier<RequestManagers> requestManagersSupplier) {
         return new CachedSupplier<ApplicationEventProcessor>() {
             @Override
             protected ApplicationEventProcessor create() {
                 RequestManagers requestManagers = requestManagersSupplier.get();
                 return new ApplicationEventProcessor(
                         logContext,
-                        applicationEventQueue,
                         requestManagers,
                         metadata
                 );

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitEvent.java

@@ -18,7 +18,6 @@
 
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.utils.Timer;
 
 import java.util.Collections;
 import java.util.Map;
@@ -30,12 +29,7 @@ public abstract class CommitEvent extends CompletableApplicationEvent<Void> {
      */
     private final Map<TopicPartition, OffsetAndMetadata> offsets;
 
-    protected CommitEvent(final Type type, final Map<TopicPartition, OffsetAndMetadata> offsets, final Timer timer) {
-        super(type, timer);
-        this.offsets = validate(offsets);
-    }
-
-    protected CommitEvent(final Type type, final Map<TopicPartition, OffsetAndMetadata> offsets, final long deadlineMs) {
+    public CommitEvent(final Type type, Map<TopicPartition, OffsetAndMetadata> offsets, final long deadlineMs) {
         super(type, deadlineMs);
         this.offsets = validate(offsets);
     }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java

@@ -16,9 +16,6 @@
  */
 package org.apache.kafka.clients.consumer.internals.events;
 
-import org.apache.kafka.common.utils.Timer;
-
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -32,13 +29,9 @@ public abstract class CompletableApplicationEvent<T> extends ApplicationEvent im
     private final CompletableFuture<T> future;
     private final long deadlineMs;
 
-    protected CompletableApplicationEvent(final Type type, final Timer timer) {
-        super(type);
-        this.future = new CompletableFuture<>();
-        Objects.requireNonNull(timer);
-        this.deadlineMs = timer.remainingMs() + timer.currentTimeMs();
-    }
-
+    /**
+     * <em>Note</em>: the {@code deadlineMs} is the future time of expiration, <em>not</em> a timeout.
+     */
     protected CompletableApplicationEvent(final Type type, final long deadlineMs) {
         super(type);
         this.future = new CompletableFuture<>();

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableBackgroundEvent.java

@@ -27,19 +27,29 @@
 public abstract class CompletableBackgroundEvent<T> extends BackgroundEvent implements CompletableEvent<T> {
 
     private final CompletableFuture<T> future;
+    private final long deadlineMs;
 
-    protected CompletableBackgroundEvent(final Type type) {
+    /**
+     * <em>Note</em>: the {@code deadlineMs} is the future time of expiration, <em>not</em> a timeout.
+     */
+    protected CompletableBackgroundEvent(final Type type, final long deadlineMs) {
         super(type);
         this.future = new CompletableFuture<>();
+        this.deadlineMs = deadlineMs;
     }
 
     @Override
     public CompletableFuture<T> future() {
         return future;
     }
 
+    @Override
+    public long deadlineMs() {
+        return deadlineMs;
+    }
+
     @Override
     protected String toStringBase() {
-        return super.toStringBase() + ", future=" + future;
+        return super.toStringBase() + ", future=" + future + ", deadlineMs=" + deadlineMs;
     }
 }