KAFKA-17439: Make polling for new records an explicit action/event in the new consumer

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

@@ -53,6 +53,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
 import org.apache.kafka.clients.consumer.internals.events.EventProcessor;
 import org.apache.kafka.clients.consumer.internals.events.FetchCommittedOffsetsEvent;
+import org.apache.kafka.clients.consumer.internals.events.FetchEvent;
 import org.apache.kafka.clients.consumer.internals.events.ListOffsetsEvent;
 import org.apache.kafka.clients.consumer.internals.events.PollEvent;
 import org.apache.kafka.clients.consumer.internals.events.ResetPositionsEvent;
@@ -696,6 +697,8 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
                 updateAssignmentMetadataIfNeeded(timer);
                 final Fetch<K, V> fetch = pollForFetches(timer);
                 if (!fetch.isEmpty()) {
+                    sendFetches(timer);
+
                     if (fetch.records().isEmpty()) {
                         log.trace("Returning empty records from `poll()` "
                             + "since the consumer's position has advanced for at least one topic partition");
@@ -1528,6 +1531,9 @@ private Fetch<K, V> pollForFetches(Timer timer) {
             return fetch;
         }
 
+        // send any new fetches (won't resend pending fetches)
+        sendFetches(timer);
+
         // We do not want to be stuck blocking in poll if we are missing some positions
         // since the offset lookup may be backing off after a failure
 
@@ -1708,6 +1714,10 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
             offsetAndMetadata.leaderEpoch().ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(topicPartition, epoch));
     }
 
+    private void sendFetches(Timer timer) {
+        applicationEventHandler.addAndGet(new FetchEvent(calculateDeadlineMs(timer)));
+    }
+
     @Override
     public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
         maybeThrowFencedInstanceException();

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

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.BiConsumer;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -41,6 +42,7 @@
 public class FetchRequestManager extends AbstractFetch implements RequestManager {
 
     private final NetworkClientDelegate networkClientDelegate;
+    private CompletableFuture<Void> pendingFetchRequestFuture;
 
     FetchRequestManager(final LogContext logContext,
                         final Time time,
@@ -65,16 +67,59 @@ protected void maybeThrowAuthFailure(Node node) {
         networkClientDelegate.maybeThrowAuthFailure(node);
     }
 
+    /**
+     * Request that a fetch request be issued to the cluster to pull down the next batch of records.
+     *
+     * <p/>
+     *
+     * The returned {@link CompletableFuture} is {@link CompletableFuture#complete(Object) completed} when the
+     * fetch request(s) have been created and enqueued into the network client's outgoing send buffer.
+     * It is <em>not completed</em> when the network client has received the data.
+     *
+     * @return Future for which the caller can wait to ensure that the requests have been enqueued
+     */
+    public CompletableFuture<Void> requestFetch() {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+
+        if (pendingFetchRequestFuture != null) {
+            // In this case, we have an outstanding fetch request, so chain the newly created future to be
+            // invoked when the outstanding fetch request is completed.
+            pendingFetchRequestFuture.whenComplete((value, exception) -> {
+                if (exception != null) {
+                    future.completeExceptionally(exception);
+                } else {
+                    future.complete(value);
+                }
+            });
+        } else {
+            pendingFetchRequestFuture = future;
+        }
+
+        return future;
+    }
+
     /**
      * {@inheritDoc}
      */
     @Override
     public PollResult poll(long currentTimeMs) {
-        return pollInternal(
+        if (pendingFetchRequestFuture == null) {
+            // If no explicit request for fetching has been issued, just short-circuit.
+            return PollResult.EMPTY;
+        }
+
+        try {
+            return pollInternal(
                 prepareFetchRequests(),
                 this::handleFetchSuccess,
                 this::handleFetchFailure
-        );
+            );
+        } finally {
+            // Completing the future here means that the caller knows that the fetch request logic has been
+            // performed. See FetchEvent for more detail.
+            pendingFetchRequestFuture.complete(null);
+            pendingFetchRequestFuture = null;
+        }
     }
 
     /**

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

@@ -32,7 +32,7 @@ public enum Type {
         COMMIT_ASYNC, COMMIT_SYNC, POLL, FETCH_COMMITTED_OFFSETS, NEW_TOPICS_METADATA_UPDATE, ASSIGNMENT_CHANGE,
         LIST_OFFSETS, RESET_POSITIONS, VALIDATE_POSITIONS, TOPIC_METADATA, ALL_TOPICS_METADATA, SUBSCRIPTION_CHANGE,
         UNSUBSCRIBE, CONSUMER_REBALANCE_LISTENER_CALLBACK_COMPLETED,
-        COMMIT_ON_CLOSE,
+        COMMIT_ON_CLOSE, FETCH,
         SHARE_FETCH, SHARE_ACKNOWLEDGE_ASYNC, SHARE_ACKNOWLEDGE_SYNC,
         SHARE_SUBSCRIPTION_CHANGE, SHARE_UNSUBSCRIBE,
         SHARE_ACKNOWLEDGE_ON_CLOSE

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

@@ -123,6 +123,10 @@ public void process(ApplicationEvent event) {
                 process((CommitOnCloseEvent) event);
                 return;
 
+            case FETCH:
+                process((FetchEvent) event);
+                return;
+
             case SHARE_FETCH:
                 process((ShareFetchEvent) event);
                 return;
@@ -161,6 +165,11 @@ private void process(final PollEvent event) {
         }
     }
 
+    private void process(final FetchEvent event) {
+        CompletableFuture<Void> future = requestManagers.fetchRequestManager.requestFetch();
+        future.whenComplete(complete(event.future()));
+    }
+
     private void process(final AsyncCommitEvent event) {
         if (!requestManagers.commitRequestManager.isPresent()) {
             return;

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

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.internals.FetchRequestManager;
+
+/**
+ * {@code FetchEvent} is sent from the consumer to signal that we want to issue a fetch request for the partitions
+ * to which the consumer is currently subscribed.
+ *
+ * <p/>
+ *
+ * <em>Note</em>: this event is completed when the {@link FetchRequestManager} has finished performing the
+ * fetch request process. It does not mean that the requests are complete. It could be the case that no fetch
+ * requests were created. Also of note is that if any fetch requests were created.
+ */
+public class FetchEvent extends CompletableApplicationEvent<Void> {
+
+    public FetchEvent(final long deadlineMs) {
+        super(Type.FETCH, deadlineMs);
+    }
+}

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -42,6 +42,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
 import org.apache.kafka.clients.consumer.internals.events.EventProcessor;
 import org.apache.kafka.clients.consumer.internals.events.FetchCommittedOffsetsEvent;
+import org.apache.kafka.clients.consumer.internals.events.FetchEvent;
 import org.apache.kafka.clients.consumer.internals.events.ListOffsetsEvent;
 import org.apache.kafka.clients.consumer.internals.events.PollEvent;
 import org.apache.kafka.clients.consumer.internals.events.ResetPositionsEvent;
@@ -1858,6 +1859,7 @@ public void testEnsurePollEventSentOnConsumerPoll() {
         consumer.subscribe(singletonList("topic1"));
         consumer.poll(Duration.ofMillis(100));
         verify(applicationEventHandler).add(any(PollEvent.class));
+        verify(applicationEventHandler).addAndGet(any(FetchEvent.class));
     }
 
     private Properties requiredConsumerConfigAndGroupId(final String groupId) {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java

@@ -3660,6 +3660,7 @@ private Fetch<K, V> collectFetch() {
         }
 
         private int sendFetches() {
+            requestFetch();
             NetworkClientDelegate.PollResult pollResult = poll(time.milliseconds());
             networkClientDelegate.addAll(pollResult.unsentRequests);
             return pollResult.unsentRequests.size();

clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

@@ -123,6 +123,7 @@ public void testApplicationEventIsProcessed(ApplicationEvent e) {
     private static Stream<Arguments> applicationEvents() {
         return Stream.of(
                 Arguments.of(new PollEvent(100)),
+                Arguments.of(new FetchEvent(calculateDeadlineMs(12345, 100))),
                 Arguments.of(new AsyncCommitEvent(new HashMap<>())),
                 Arguments.of(new SyncCommitEvent(new HashMap<>(), 500)),
                 Arguments.of(new ResetPositionsEvent(500)),