KAFKA-14960: [Part I]TopicMetadataRequestManager Implementation

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletableRequest.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;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A request that is completable by the user. This is used to wrap requests that are sent to the background thread
+ * for completion.
+ * @param <T> The type of the response.
+ */
+public class CompletableRequest<T> {
+    private final CompletableFuture<T> future;
+
+    public CompletableRequest() {
+        this.future = new CompletableFuture<>();
+    }
+
+    public CompletableFuture<T> future() {
+        return this.future;
+    }
+}

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

@@ -87,7 +87,8 @@ public class DefaultBackgroundThread extends KafkaThread {
                             final ApplicationEventProcessor processor,
                             final CoordinatorRequestManager coordinatorManager,
                             final CommitRequestManager commitRequestManager,
-                            final ListOffsetsRequestManager listOffsetsRequestManager) {
+                            final ListOffsetsRequestManager listOffsetsRequestManager,
+                            final TopicMetadataRequestManager topicMetadataRequestManager) {
         super(BACKGROUND_THREAD_NAME, true);
         this.time = time;
         this.log = logContext.logger(getClass());
@@ -103,6 +104,7 @@ public class DefaultBackgroundThread extends KafkaThread {
 
         this.requestManagers = new RequestManagers(
                 listOffsetsRequestManager,
+                topicMetadataRequestManager,
                 Optional.ofNullable(coordinatorManager),
                 Optional.ofNullable(commitRequestManager));
     }
@@ -129,55 +131,57 @@ public DefaultBackgroundThread(final Time time,
             this.metadata = metadata;
 
             final NetworkClient networkClient = ClientUtils.createNetworkClient(config,
-                    metrics,
-                    CONSUMER_METRIC_GROUP_PREFIX,
-                    logContext,
-                    apiVersions,
-                    time,
-                    CONSUMER_MAX_INFLIGHT_REQUESTS_PER_CONNECTION,
-                    metadata,
-                    fetcherThrottleTimeSensor);
+                metrics,
+                CONSUMER_METRIC_GROUP_PREFIX,
+                logContext,
+                apiVersions,
+                time,
+                CONSUMER_MAX_INFLIGHT_REQUESTS_PER_CONNECTION,
+                metadata,
+                fetcherThrottleTimeSensor);
 
             this.networkClientDelegate = new NetworkClientDelegate(this.time, this.config, logContext, networkClient);
             this.errorEventHandler = new ErrorEventHandler(this.backgroundEventQueue);
             this.groupState = new GroupState(rebalanceConfig);
             long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
 
             ListOffsetsRequestManager offsetsRequestManager =
-                    new ListOffsetsRequestManager(
-                            subscriptionState,
-                            metadata,
-                            getConfiguredIsolationLevel(config),
-                            time,
-                            apiVersions,
-                            logContext);
+                new ListOffsetsRequestManager(
+                    subscriptionState,
+                    metadata,
+                    getConfiguredIsolationLevel(config),
+                    time,
+                    apiVersions,
+                    logContext);
             CoordinatorRequestManager coordinatorRequestManager = null;
             CommitRequestManager commitRequestManager = null;
+            TopicMetadataRequestManager topicMetadataRequestManger = new TopicMetadataRequestManager(
+                logContext,
+                config);
 
             if (groupState.groupId != null) {
                 coordinatorRequestManager = new CoordinatorRequestManager(this.time,
-                        logContext,
-                        retryBackoffMs,
-                        this.errorEventHandler,
-                        groupState.groupId);
+                    logContext,
+                    retryBackoffMs, // TODO: let's pass in the config directly
+                    this.errorEventHandler,
+                    groupState.groupId);
                 commitRequestManager = new CommitRequestManager(this.time,
-                        logContext,
-                        subscriptions,
-                        config,
-                        coordinatorRequestManager,
-                        groupState);
+                    logContext,
+                    subscriptions,
+                    config,
+                    coordinatorRequestManager,
+                    groupState);
             }
 
             this.requestManagers = new RequestManagers(
-                    offsetsRequestManager,
-                    Optional.ofNullable(coordinatorRequestManager),
-                    Optional.ofNullable(commitRequestManager));
-
+                offsetsRequestManager,
+                topicMetadataRequestManger,
+                Optional.ofNullable(coordinatorRequestManager),
+                Optional.ofNullable(commitRequestManager));
             this.applicationEventProcessor = new ApplicationEventProcessor(
-                    backgroundEventQueue,
-                    requestManagers,
-                    metadata);
-
+                backgroundEventQueue,
+                requestManagers,
+                metadata);
         } catch (final Exception e) {
             close();
             throw new KafkaException("Failed to construct background processor", e.getCause());
@@ -204,10 +208,10 @@ public void run() {
             }
         } catch (final Throwable t) {
             log.error("The background thread failed due to unexpected error", t);
-            throw new RuntimeException(t);
+            throw new KafkaException(t);
         } finally {
             close();
-            log.debug("Exited run loop");
+            log.debug("Background thread closed");
         }
     }
 

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

@@ -40,6 +40,7 @@
 import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 /**
  * A wrapper around the {@link org.apache.kafka.clients.NetworkClient} to handle network poll and send operations.
@@ -199,20 +200,21 @@ public PollResult(final long timeMsTillNextPoll, final List<UnsentRequest> unsen
     public static class UnsentRequest {
         private final AbstractRequest.Builder<?> requestBuilder;
         private final FutureCompletionHandler handler;
-        private Optional<Node> node; // empty if random node can be choosen
+        private final Optional<Node> node; // empty if random node can be chosen
         private Timer timer;
 
         public UnsentRequest(final AbstractRequest.Builder<?> requestBuilder, final Optional<Node> node) {
-            this(requestBuilder, node, new FutureCompletionHandler());
+            Objects.requireNonNull(requestBuilder);
+            this.requestBuilder = requestBuilder;
+            this.node = node;
+            this.handler = new FutureCompletionHandler();
         }
 
         public UnsentRequest(final AbstractRequest.Builder<?> requestBuilder,
                              final Optional<Node> node,
-                             final FutureCompletionHandler handler) {
-            Objects.requireNonNull(requestBuilder);
-            this.requestBuilder = requestBuilder;
-            this.node = node;
-            this.handler = handler;
+                             final BiConsumer<ClientResponse, Throwable> callback) {
+            this(requestBuilder, node);
+            this.handler.future.whenComplete(callback);
         }
 
         public void setTimer(final Time time, final long requestTimeoutMs) {
@@ -249,10 +251,6 @@ public void onFailure(final RuntimeException e) {
             future.completeExceptionally(e);
         }
 
-        public CompletableFuture<ClientResponse> future() {
-            return future;
-        }
-
         @Override
         public void onComplete(final ClientResponse response) {
             if (response.authenticationException() != null) {
@@ -266,5 +264,4 @@ public void onComplete(final ClientResponse response) {
             }
         }
     }
-
 }

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

@@ -33,21 +33,23 @@ public class RequestManagers {
     public final Optional<CoordinatorRequestManager> coordinatorRequestManager;
     public final Optional<CommitRequestManager> commitRequestManager;
     public final ListOffsetsRequestManager listOffsetsRequestManager;
+    public final TopicMetadataRequestManager topicMetadataRequestManager;
     private final List<Optional<? extends RequestManager>> entries;
 
     public RequestManagers(ListOffsetsRequestManager listOffsetsRequestManager,
+                           TopicMetadataRequestManager topicMetadataRequestManager,
                            Optional<CoordinatorRequestManager> coordinatorRequestManager,
                            Optional<CommitRequestManager> commitRequestManager) {
-        this.listOffsetsRequestManager = requireNonNull(listOffsetsRequestManager,
-                "ListOffsetsRequestManager cannot be null");
+        this.listOffsetsRequestManager = requireNonNull(listOffsetsRequestManager, "ListOffsetsRequestManager cannot be null");
         this.coordinatorRequestManager = coordinatorRequestManager;
         this.commitRequestManager = commitRequestManager;
-
+        this.topicMetadataRequestManager = topicMetadataRequestManager;
 
         List<Optional<? extends RequestManager>> list = new ArrayList<>();
         list.add(coordinatorRequestManager);
         list.add(commitRequestManager);
         list.add(Optional.of(listOffsetsRequestManager));
+        list.add(Optional.of(topicMetadataRequestManager));
         entries = Collections.unmodifiableList(list);
     }