KAFKA-14274 [6, 7]: Introduction of fetch request manager

checkstyle/import-control.xml

@@ -205,6 +205,10 @@
 
     <subpackage name="consumer">
       <allow pkg="org.apache.kafka.clients.consumer" />
+
+      <subpackage name="internals">
+        <allow pkg="org.apache.kafka.clients" />
+      </subpackage>
     </subpackage>
 
     <subpackage name="producer">

checkstyle/suppressions.xml

@@ -61,14 +61,16 @@
               files="AbstractRequest.java"/>
     <suppress checks="ClassFanOutComplexity"
               files="AbstractResponse.java"/>
+    <suppress checks="ClassFanOutComplexity"
+              files="PrototypeAsyncConsumer.java"/>
 
     <suppress checks="MethodLength"
               files="(KerberosLogin|RequestResponseTest|ConnectMetricsRegistry|KafkaConsumer|AbstractStickyAssignor).java"/>
 
     <suppress checks="ParameterNumber"
               files="(NetworkClient|FieldSpec|KafkaRaftClient).java"/>
     <suppress checks="ParameterNumber"
-              files="(KafkaConsumer|ConsumerCoordinator).java"/>
+              files="(KafkaConsumer|PrototypeAsyncConsumer|ConsumerCoordinator).java"/>
     <suppress checks="ParameterNumber"
               files="(RecordAccumulator|Sender).java"/>
     <suppress checks="ParameterNumber"
@@ -79,7 +81,7 @@
               files="MemoryRecordsBuilder.java"/>
 
     <suppress checks="ClassDataAbstractionCoupling"
-              files="(KafkaConsumer|ConsumerCoordinator|AbstractFetch|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|Admin|KafkaAdminClient|MockAdminClient|KafkaRaftClient|KafkaRaftClientTest).java"/>
+              files="(KafkaConsumer|PrototypeAsyncConsumer|ConsumerCoordinator|AbstractFetch|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|Admin|KafkaAdminClient|MockAdminClient|KafkaRaftClient|KafkaRaftClientTest).java"/>
     <suppress checks="ClassDataAbstractionCoupling"
               files="(Errors|SaslAuthenticatorTest|AgentTest|CoordinatorTest).java"/>
 
@@ -108,10 +110,10 @@
 
     <!-- Clients tests -->
     <suppress checks="ClassDataAbstractionCoupling"
-              files="(Sender|Fetcher|OffsetFetcher|KafkaConsumer|Metrics|RequestResponse|TransactionManager|KafkaAdminClient|Message|KafkaProducer)Test.java"/>
+              files="(Sender|Fetcher|FetchRequestManager|OffsetFetcher|KafkaConsumer|PrototypeAsyncConsumer|Metrics|RequestResponse|TransactionManager|KafkaAdminClient|Message|KafkaProducer)Test.java"/>
 
     <suppress checks="ClassFanOutComplexity"
-              files="(ConsumerCoordinator|KafkaConsumer|RequestResponse|Fetcher|KafkaAdminClient|Message|KafkaProducer)Test.java"/>
+              files="(ConsumerCoordinator|KafkaConsumer|RequestResponse|Fetcher|FetchRequestManager|KafkaAdminClient|Message|KafkaProducer)Test.java"/>
 
     <suppress checks="ClassFanOutComplexity"
               files="MockAdminClient.java"/>
@@ -120,7 +122,7 @@
               files="(OffsetFetcher|RequestResponse)Test.java"/>
 
     <suppress checks="JavaNCSS"
-              files="RequestResponseTest.java|FetcherTest.java|KafkaAdminClientTest.java"/>
+              files="RequestResponseTest.java|FetcherTest.java|FetchRequestManagerTest.java|KafkaAdminClientTest.java"/>
 
     <suppress checks="NPathComplexity"
               files="MemoryRecordsTest|MetricsTest|RequestResponseTest|TestSslUtils|AclAuthorizerBenchmark"/>

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -752,13 +752,14 @@ public KafkaConsumer(Map<String, Object> configs,
                         config.getBoolean(ConsumerConfig.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED),
                         config.getString(ConsumerConfig.CLIENT_RACK_CONFIG));
             }
-            FetchConfig<K, V> fetchConfig = createFetchConfig(config, this.deserializers);
+            FetchConfig fetchConfig = createFetchConfig(config);
             this.fetcher = new Fetcher<>(
                     logContext,
                     this.client,
                     this.metadata,
                     this.subscriptions,
                     fetchConfig,
+                    this.deserializers,
                     fetchMetricsManager,
                     this.time);
             this.offsetFetcher = new OffsetFetcher(logContext,

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

@@ -18,10 +18,13 @@
 
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.clients.FetchSessionHandler;
+import org.apache.kafka.clients.KafkaClient;
+import org.apache.kafka.clients.NetworkClientUtils;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.internals.IdempotentCloser;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -37,11 +40,9 @@
 import org.slf4j.helpers.MessageFormatter;
 
 import java.io.Closeable;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -51,17 +52,14 @@
 
 /**
  * {@code AbstractFetch} represents the basic state and logic for record fetching processing.
- * @param <K> Type for the message key
- * @param <V> Type for the message value
  */
-public abstract class AbstractFetch<K, V> implements Closeable {
+public abstract class AbstractFetch implements Closeable {
 
     private final Logger log;
     protected final LogContext logContext;
-    protected final ConsumerNetworkClient client;
     protected final ConsumerMetadata metadata;
     protected final SubscriptionState subscriptions;
-    protected final FetchConfig<K, V> fetchConfig;
+    protected final FetchConfig fetchConfig;
     protected final Time time;
     protected final FetchMetricsManager metricsManager;
     protected final FetchBuffer fetchBuffer;
@@ -72,15 +70,13 @@ public abstract class AbstractFetch<K, V> implements Closeable {
     private final Map<Integer, FetchSessionHandler> sessionHandlers;
 
     public AbstractFetch(final LogContext logContext,
-                         final ConsumerNetworkClient client,
                          final ConsumerMetadata metadata,
                          final SubscriptionState subscriptions,
-                         final FetchConfig<K, V> fetchConfig,
+                         final FetchConfig fetchConfig,
                          final FetchMetricsManager metricsManager,
                          final Time time) {
         this.log = logContext.logger(AbstractFetch.class);
         this.logContext = logContext;
-        this.client = client;
         this.metadata = metadata;
         this.subscriptions = subscriptions;
         this.fetchConfig = fetchConfig;
@@ -92,6 +88,23 @@ public AbstractFetch(final LogContext logContext,
         this.time = time;
     }
 
+    /**
+     * Check if the node is disconnected and unavailable for immediate reconnection (i.e. if it is in
+     * reconnect backoff window following the disconnect).
+     *
+     * @param node {@link Node} to check for availability
+     * @see NetworkClientUtils#isUnavailable(KafkaClient, Node, Time)
+     */
+    protected abstract boolean isUnavailable(Node node);
+
+    /**
+     * Checks for an authentication error on a given node and throws the exception if it exists.
+     *
+     * @param node {@link Node} to check for a previous {@link AuthenticationException}; if found it is thrown
+     * @see NetworkClientUtils#maybeThrowAuthFailure(KafkaClient, Node)
+     */
+    protected abstract void maybeThrowAuthFailure(Node node);
+
     /**
      * Return whether we have any completed fetches pending return to the user. This method is thread-safe. Has
      * visibility for testing.
@@ -317,7 +330,7 @@ Node selectReadReplica(final TopicPartition partition, final Node leaderReplica,
         }
     }
 
-    private Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
+    protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
         final Cluster cluster = metadata.fetch();
         Map<Node, FetchSessionHandler.Builder> fetchable = new LinkedHashMap<>();
 
@@ -330,7 +343,7 @@ private Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSession
                 // skip sending the close request.
                 final Node fetchTarget = cluster.nodeById(fetchTargetNodeId);
 
-                if (fetchTarget == null || client.isUnavailable(fetchTarget)) {
+                if (fetchTarget == null || isUnavailable(fetchTarget)) {
                     log.debug("Skip sending close session request to broker {} since it is not reachable", fetchTarget);
                     return;
                 }
@@ -377,8 +390,8 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
             // Use the preferred read replica if set, otherwise the partition's leader
             Node node = selectReadReplica(partition, leaderOpt.get(), currentTimeMs);
 
-            if (client.isUnavailable(node)) {
-                client.maybeThrowAuthFailure(node);
+            if (isUnavailable(node)) {
+                maybeThrowAuthFailure(node);
 
                 // If we try to send during the reconnect backoff window, then the request is just
                 // going to be failed anyway before being sent, so skip sending the request for now
@@ -412,46 +425,6 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
         return reqs;
     }
 
-    protected void maybeCloseFetchSessions(final Timer timer) {
-        final List<RequestFuture<ClientResponse>> requestFutures = new ArrayList<>();
-        Map<Node, FetchSessionHandler.FetchRequestData> fetchRequestMap = prepareCloseFetchSessionRequests();
-
-        for (Map.Entry<Node, FetchSessionHandler.FetchRequestData> entry : fetchRequestMap.entrySet()) {
-            final Node fetchTarget = entry.getKey();
-            final FetchSessionHandler.FetchRequestData data = entry.getValue();
-            final FetchRequest.Builder request = createFetchRequest(fetchTarget, data);
-            final RequestFuture<ClientResponse> responseFuture = client.send(fetchTarget, request);
-
-            responseFuture.addListener(new RequestFutureListener<ClientResponse>() {
-                @Override
-                public void onSuccess(ClientResponse value) {
-                    handleCloseFetchSessionResponse(fetchTarget, data);
-                }
-
-                @Override
-                public void onFailure(RuntimeException e) {
-                    handleCloseFetchSessionResponse(fetchTarget, data, e);
-                }
-            });
-
-            requestFutures.add(responseFuture);
-        }
-
-        // Poll to ensure that request has been written to the socket. Wait until either the timer has expired or until
-        // all requests have received a response.
-        while (timer.notExpired() && !requestFutures.stream().allMatch(RequestFuture::isDone)) {
-            client.poll(timer, null, true);
-        }
-
-        if (!requestFutures.stream().allMatch(RequestFuture::isDone)) {
-            // we ran out of time before completing all futures. It is ok since we don't want to block the shutdown
-            // here.
-            log.debug("All requests couldn't be sent in the specific timeout period {}ms. " +
-                    "This may result in unnecessary fetch sessions at the broker. Consider increasing the timeout passed for " +
-                    "KafkaConsumer.close(Duration timeout)", timer.timeoutMs());
-        }
-    }
-
     // Visible for testing
     protected FetchSessionHandler sessionHandler(int node) {
         return sessionHandlers.get(node);
@@ -467,8 +440,6 @@ protected FetchSessionHandler sessionHandler(int node) {
     // Visible for testing
     protected void closeInternal(Timer timer) {
         // we do not need to re-enable wake-ups since we are closing already
-        client.disableWakeups();
-        maybeCloseFetchSessions(timer);
         Utils.closeQuietly(fetchBuffer, "fetchBuffer");
         Utils.closeQuietly(decompressionBufferSupplier, "decompressionBufferSupplier");
     }

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

@@ -0,0 +1,43 @@
+/*
+ * 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;
+
+import java.util.function.Supplier;
+
+/**
+ * Simple {@link Supplier} that caches the initial creation of the object and stores it for later calls
+ * to {@link #get()}.
+ *
+ * <p/>
+ *
+ * <em>Note</em>: this class is not thread safe! Use only in contexts which are designed/guaranteed to be
+ * single-threaded.
+ */
+public abstract class CachedSupplier<T> implements Supplier<T> {
+
+    private T result;
+
+    protected abstract T create();
+
+    @Override
+    public T get() {
+        if (result == null)
+            result = create();
+
+        return result;
+    }
+}

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

@@ -53,7 +53,7 @@
 /**
  * {@link CompletedFetch} represents a {@link RecordBatch batch} of {@link Record records} that was returned from the
  * broker via a {@link FetchRequest}. It contains logic to maintain state between calls to
- * {@link #fetchRecords(FetchConfig, int)}.
+ * {@link #fetchRecords(FetchConfig, Deserializers, int)}.
  */
 public class CompletedFetch {
 
@@ -135,7 +135,8 @@ void recordAggregatedMetrics(int bytes, int records) {
     /**
      * Draining a {@link CompletedFetch} will signal that the data has been consumed and the underlying resources
      * are closed. This is somewhat analogous to {@link Closeable#close() closing}, though no error will result if a
-     * caller invokes {@link #fetchRecords(FetchConfig, int)}; an empty {@link List list} will be returned instead.
+     * caller invokes {@link #fetchRecords(FetchConfig, Deserializers, int)}; an empty {@link List list} will be
+     * returned instead.
      */
     void drain() {
         if (!isConsumed) {
@@ -151,7 +152,7 @@ void drain() {
         }
     }
 
-    private <K, V> void maybeEnsureValid(FetchConfig<K, V> fetchConfig, RecordBatch batch) {
+    private void maybeEnsureValid(FetchConfig fetchConfig, RecordBatch batch) {
         if (fetchConfig.checkCrcs && batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
             try {
                 batch.ensureValid();
@@ -162,7 +163,7 @@ private <K, V> void maybeEnsureValid(FetchConfig<K, V> fetchConfig, RecordBatch
         }
     }
 
-    private <K, V> void maybeEnsureValid(FetchConfig<K, V> fetchConfig, Record record) {
+    private void maybeEnsureValid(FetchConfig fetchConfig, Record record) {
         if (fetchConfig.checkCrcs) {
             try {
                 record.ensureValid();
@@ -180,7 +181,7 @@ private void maybeCloseRecordStream() {
         }
     }
 
-    private <K, V> Record nextFetchedRecord(FetchConfig<K, V> fetchConfig) {
+    private Record nextFetchedRecord(FetchConfig fetchConfig) {
         while (true) {
             if (records == null || !records.hasNext()) {
                 maybeCloseRecordStream();
@@ -249,7 +250,9 @@ private <K, V> Record nextFetchedRecord(FetchConfig<K, V> fetchConfig) {
      * @param maxRecords The number of records to return; the number returned may be {@code 0 <= maxRecords}
      * @return {@link ConsumerRecord Consumer records}
      */
-    <K, V> List<ConsumerRecord<K, V>> fetchRecords(FetchConfig<K, V> fetchConfig, int maxRecords) {
+    <K, V> List<ConsumerRecord<K, V>> fetchRecords(FetchConfig fetchConfig,
+                                                   Deserializers<K, V> deserializers,
+                                                   int maxRecords) {
         // Error when fetching the next record before deserialization.
         if (corruptLastRecord)
             throw new KafkaException("Received exception when fetching the next record from " + partition
@@ -276,7 +279,7 @@ <K, V> List<ConsumerRecord<K, V>> fetchRecords(FetchConfig<K, V> fetchConfig, in
 
                 Optional<Integer> leaderEpoch = maybeLeaderEpoch(currentBatch.partitionLeaderEpoch());
                 TimestampType timestampType = currentBatch.timestampType();
-                ConsumerRecord<K, V> record = parseRecord(fetchConfig, partition, leaderEpoch, timestampType, lastRecord);
+                ConsumerRecord<K, V> record = parseRecord(deserializers, partition, leaderEpoch, timestampType, lastRecord);
                 records.add(record);
                 recordsRead++;
                 bytesRead += lastRecord.sizeInBytes();
@@ -302,7 +305,7 @@ <K, V> List<ConsumerRecord<K, V>> fetchRecords(FetchConfig<K, V> fetchConfig, in
     /**
      * Parse the record entry, deserializing the key / value fields if necessary
      */
-    <K, V> ConsumerRecord<K, V> parseRecord(FetchConfig<K, V> fetchConfig,
+    <K, V> ConsumerRecord<K, V> parseRecord(Deserializers<K, V> deserializers,
                                             TopicPartition partition,
                                             Optional<Integer> leaderEpoch,
                                             TimestampType timestampType,
@@ -312,16 +315,16 @@ <K, V> ConsumerRecord<K, V> parseRecord(FetchConfig<K, V> fetchConfig,
             long timestamp = record.timestamp();
             Headers headers = new RecordHeaders(record.headers());
             ByteBuffer keyBytes = record.key();
-            K key = keyBytes == null ? null : fetchConfig.deserializers.keyDeserializer.deserialize(partition.topic(), headers, keyBytes);
+            K key = keyBytes == null ? null : deserializers.keyDeserializer.deserialize(partition.topic(), headers, keyBytes);
             ByteBuffer valueBytes = record.value();
-            V value = valueBytes == null ? null : fetchConfig.deserializers.valueDeserializer.deserialize(partition.topic(), headers, valueBytes);
+            V value = valueBytes == null ? null : deserializers.valueDeserializer.deserialize(partition.topic(), headers, valueBytes);
             return new ConsumerRecord<>(partition.topic(), partition.partition(), offset,
                     timestamp, timestampType,
                     keyBytes == null ? ConsumerRecord.NULL_SIZE : keyBytes.remaining(),
                     valueBytes == null ? ConsumerRecord.NULL_SIZE : valueBytes.remaining(),
                     key, value, headers, leaderEpoch);
         } catch (RuntimeException e) {
-            log.error("Deserializers with error: {}", fetchConfig.deserializers);
+            log.error("Deserializers with error: {}", deserializers);
             throw new RecordDeserializationException(partition, record.offset(),
                     "Error deserializing key/value for partition " + partition +
                             " at offset " + record.offset() + ". If needed, please seek past the record to continue consumption.", e);

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

@@ -142,10 +142,9 @@ public static FetchMetricsManager createFetchMetricsManager(Metrics metrics) {
         return new FetchMetricsManager(metrics, metricsRegistry);
     }
 
-    public static <K, V> FetchConfig<K, V> createFetchConfig(ConsumerConfig config,
-                                                             Deserializers<K, V> deserializers) {
+    public static FetchConfig createFetchConfig(ConsumerConfig config) {
         IsolationLevel isolationLevel = configuredIsolationLevel(config);
-        return new FetchConfig<>(config, deserializers, isolationLevel);
+        return new FetchConfig(config, isolationLevel);
     }
 
     @SuppressWarnings("unchecked")