diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java index 82c22505ddf40..c94966389c254 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.clients.consumer.internals; -import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.RetriableCommitFailedException; @@ -98,6 +97,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList()); } + pendingRequests.inflightOffsetFetches.forEach(System.out::println); return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.unmodifiableList(pendingRequests.drain(currentTimeMs))); } @@ -120,15 +120,15 @@ private void maybeAutoCommit() { /** * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an - * {@link OffsetCommitRequestState} and enqueue it to send later. + * {@link CompletableOffsetCommitRequest} and enqueue it to send later. */ - public CompletableFuture addOffsetCommitRequest(final Map offsets) { + public CompletableFuture addOffsetCommitRequest(final Map offsets) { return pendingRequests.addOffsetCommitRequest(offsets); } /** * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an - * {@link OffsetFetchRequestState} and enqueue it to send later. + * {@link CompletableOffsetFetchRequest} and enqueue it to send later. */ public CompletableFuture> addOffsetFetchRequest(final Set partitions) { return pendingRequests.addOffsetFetchRequest(partitions); @@ -140,17 +140,17 @@ public void updateAutoCommitTimer(final long currentTimeMs) { // Visible for testing - List unsentOffsetFetchRequests() { + List unsentOffsetFetchRequests() { return pendingRequests.unsentOffsetFetches; } // Visible for testing - Queue unsentOffsetCommitRequests() { + Queue unsentOffsetCommitRequests() { return pendingRequests.unsentOffsetCommits; } // Visible for testing - CompletableFuture sendAutoCommit(final Map allConsumedOffsets) { + CompletableFuture sendAutoCommit(final Map allConsumedOffsets) { log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets); return this.addOffsetCommitRequest(allConsumedOffsets) .whenComplete((response, throwable) -> { @@ -170,28 +170,22 @@ CompletableFuture sendAutoCommit(final Map { private final Map offsets; private final String groupId; private final GroupState.Generation generation; private final String groupInstanceId; - private final NetworkClientDelegate.FutureCompletionHandler future; - public OffsetCommitRequestState(final Map offsets, - final String groupId, - final String groupInstanceId, - final GroupState.Generation generation) { + public CompletableOffsetCommitRequest(final Map offsets, + final String groupId, + final String groupInstanceId, + final GroupState.Generation generation) { this.offsets = offsets; - this.future = new NetworkClientDelegate.FutureCompletionHandler(); this.groupId = groupId; this.generation = generation; this.groupInstanceId = groupInstanceId; } - public CompletableFuture future() { - return future.future(); - } - public NetworkClientDelegate.UnsentRequest toUnsentRequest() { Map requestTopicDataMap = new HashMap<>(); for (Map.Entry entry : offsets.entrySet()) { @@ -223,25 +217,30 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() { return new NetworkClientDelegate.UnsentRequest( builder, coordinatorRequestManager.coordinator(), - future); + (response, throwable) -> { + if (throwable == null) { + this.future().complete(null); + } else { + this.future().completeExceptionally(throwable); + } + }); } } - private class OffsetFetchRequestState extends RequestState { + private class CompletableOffsetFetchRequest extends CompletableRequest> { public final Set requestedPartitions; public final GroupState.Generation requestedGeneration; - public CompletableFuture> future; + private final RequestState requestState; - public OffsetFetchRequestState(final Set partitions, - final GroupState.Generation generation, - final long retryBackoffMs) { - super(logContext, retryBackoffMs); + public CompletableOffsetFetchRequest(final Set partitions, + final GroupState.Generation generation, + final long retryBackoffMs) { this.requestedPartitions = partitions; this.requestedGeneration = generation; - this.future = new CompletableFuture<>(); + this.requestState = new RequestState(logContext, retryBackoffMs); } - public boolean sameRequest(final OffsetFetchRequestState request) { + public boolean sameRequest(final CompletableOffsetFetchRequest request) { return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions); } @@ -251,13 +250,10 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTim true, new ArrayList<>(this.requestedPartitions), throwOnFetchStableOffsetUnsupported); - NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest( + return new NetworkClientDelegate.UnsentRequest( builder, - coordinatorRequestManager.coordinator()); - unsentRequest.future().whenComplete((r, t) -> { - onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody()); - }); - return unsentRequest; + coordinatorRequestManager.coordinator(), + (r, t) -> onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody())); } public void onResponse( @@ -283,15 +279,15 @@ private void onFailure(final long currentTimeMs, coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds()); retry(currentTimeMs); } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) { - future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId)); + future().completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId)); } else { - future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message())); + future().completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message())); } } private void retry(final long currentTimeMs) { - onFailedAttempt(currentTimeMs); - onSendAttempt(currentTimeMs); + this.requestState.onFailedAttempt(currentTimeMs); + this.requestState.onSendAttempt(currentTimeMs); pendingRequests.addOffsetFetchRequest(this); } @@ -310,7 +306,7 @@ private void onSuccess(final long currentTimeMs, log.debug("Failed to fetch offset for partition {}: {}", tp, error.message()); if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) { - future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " + + future().completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " + "not " + "exist")); return; @@ -322,7 +318,7 @@ private void onSuccess(final long currentTimeMs, } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) { unstableTxnOffsetTopicPartitions.add(tp); } else { - future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " + + future().completeExceptionally(new KafkaException("Unexpected error in fetch offset " + "response for partition " + tp + ": " + error.message())); return; } @@ -337,7 +333,7 @@ private void onSuccess(final long currentTimeMs, } if (unauthorizedTopics != null) { - future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics)); + future().completeExceptionally(new TopicAuthorizationException(unauthorizedTopics)); } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) { // TODO: Optimization question: Do we need to retry all partitions upon a single partition error? log.info("The following partitions still have unstable offsets " + @@ -347,43 +343,43 @@ private void onSuccess(final long currentTimeMs, "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions); retry(currentTimeMs); } else { - future.complete(offsets); + future().complete(offsets); } } - private CompletableFuture> chainFuture(final CompletableFuture> future) { - return this.future.whenComplete((r, t) -> { + private CompletableFuture> chainFuture(final CompletableFuture> otherFuture) { + return this.future().whenComplete((r, t) -> { if (t != null) { - future.completeExceptionally(t); + otherFuture.completeExceptionally(t); } else { - future.complete(r); + otherFuture.complete(r); } }); } } /** - *

This is used to stage the unsent {@link OffsetCommitRequestState} and {@link OffsetFetchRequestState}. + *

This is used to stage the unsent {@link CompletableOffsetCommitRequest} and {@link CompletableOffsetFetchRequest}. *

  • unsentOffsetCommits holds the offset commit requests that have not been sent out *
  • unsentOffsetFetches holds the offset fetch requests that have not been sent out
    • - *
  • inflightOffsetFetches holds the offset fetch requests that have been sent out but incompleted. - * + *
  • inflightOffsetFetches holds the offset fetch requests that have been sent out but not completed. + *

    * {@code addOffsetFetchRequest} dedupes the requests to avoid sending the same requests. */ class PendingRequests { // Queue is used to ensure the sequence of commit - Queue unsentOffsetCommits = new LinkedList<>(); - List unsentOffsetFetches = new ArrayList<>(); - List inflightOffsetFetches = new ArrayList<>(); + Queue unsentOffsetCommits = new LinkedList<>(); + List unsentOffsetFetches = new ArrayList<>(); + List inflightOffsetFetches = new ArrayList<>(); - public boolean hasUnsentRequests() { + boolean hasUnsentRequests() { return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty(); } - public CompletableFuture addOffsetCommitRequest(final Map offsets) { + CompletableFuture addOffsetCommitRequest(final Map offsets) { // TODO: Dedupe committing the same offsets to the same partitions - OffsetCommitRequestState request = new OffsetCommitRequestState( + CompletableOffsetCommitRequest request = new CompletableOffsetCommitRequest( offsets, groupState.groupId, groupState.groupInstanceId.orElse(null), @@ -393,24 +389,24 @@ public CompletableFuture addOffsetCommitRequest(final MapAdding an offset fetch request to the outgoing buffer. If the same request was made, we chain the future - * to the existing one. + *

    Adding an offset fetch request to the outgoing buffer. If the same request was made, we chain the future + * to the existing one. * - *

    If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches} - * upon completion. + *

    If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches} + * upon completion. */ - private CompletableFuture> addOffsetFetchRequest(final OffsetFetchRequestState request) { - Optional dupe = + private CompletableFuture> addOffsetFetchRequest(final CompletableOffsetFetchRequest request) { + Optional dupe = unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny(); - Optional inflight = + Optional inflight = inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny(); if (dupe.isPresent() || inflight.isPresent()) { log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions); - dupe.orElseGet(() -> inflight.get()).chainFuture(request.future); + dupe.orElseGet(() -> inflight.get()).chainFuture(request.future()); } else { // remove the request from the outbound buffer: inflightOffsetFetches - request.future.whenComplete((r, t) -> { + request.future().whenComplete((r, t) -> { if (!inflightOffsetFetches.remove(request)) { log.warn("A duplicated, inflight, request was identified, but unable to find it in the " + "outbound buffer:" + request); @@ -418,11 +414,11 @@ private CompletableFuture> addOffsetFetch }); this.unsentOffsetFetches.add(request); } - return request.future; + return request.future(); } private CompletableFuture> addOffsetFetchRequest(final Set partitions) { - OffsetFetchRequestState request = new OffsetFetchRequestState( + CompletableOffsetFetchRequest request = new CompletableOffsetFetchRequest( partitions, groupState.generation, retryBackoffMs); @@ -432,27 +428,27 @@ private CompletableFuture> addOffsetFetch /** * Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the * {@code inflightOffsetFetches} to bookkeep all of the inflight requests. - * + *

    * Note: Sendable requests are determined by their timer as we are expecting backoff on failed attempt. See * {@link RequestState}. **/ - public List drain(final long currentTimeMs) { + List drain(final long currentTimeMs) { List unsentRequests = new ArrayList<>(); // Add all unsent offset commit requests to the unsentRequests list unsentRequests.addAll( unsentOffsetCommits.stream() - .map(OffsetCommitRequestState::toUnsentRequest) + .map(CompletableOffsetCommitRequest::toUnsentRequest) .collect(Collectors.toList())); // Partition the unsent offset fetch requests into sendable and non-sendable lists - Map> partitionedBySendability = + Map> partitionedBySendability = unsentOffsetFetches.stream() - .collect(Collectors.partitioningBy(request -> request.canSendRequest(currentTimeMs))); + .collect(Collectors.partitioningBy(request -> request.requestState.canSendRequest(currentTimeMs))); // Add all sendable offset fetch requests to the unsentRequests list and to the inflightOffsetFetches list - for (OffsetFetchRequestState request : partitionedBySendability.get(true)) { - request.onSendAttempt(currentTimeMs); + for (CompletableOffsetFetchRequest request : partitionedBySendability.get(true)) { + request.requestState.onSendAttempt(currentTimeMs); unsentRequests.add(request.toUnsentRequest(currentTimeMs)); inflightOffsetFetches.add(request); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletableRequest.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletableRequest.java new file mode 100644 index 0000000000000..bbfc12abf5a7b --- /dev/null +++ b/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 The type of the response. + */ +public class CompletableRequest { + private final CompletableFuture future; + + public CompletableRequest() { + this.future = new CompletableFuture<>(); + } + + public CompletableFuture future() { + return this.future; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java index 1a0a73cb349c6..2ddd81a0a9934 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java +++ b/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,14 +131,14 @@ 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); @@ -144,40 +146,42 @@ public DefaultBackgroundThread(final Time time, 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"); } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java index eb4d6800a9925..102c16cfcb155 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java +++ b/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 unsen public static class UnsentRequest { private final AbstractRequest.Builder requestBuilder; private final FutureCompletionHandler handler; - private Optional node; // empty if random node can be choosen + private final Optional node; // empty if random node can be chosen private Timer timer; public UnsentRequest(final AbstractRequest.Builder requestBuilder, final Optional 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, - final FutureCompletionHandler handler) { - Objects.requireNonNull(requestBuilder); - this.requestBuilder = requestBuilder; - this.node = node; - this.handler = handler; + final BiConsumer 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 future() { - return future; - } - @Override public void onComplete(final ClientResponse response) { if (response.authenticationException() != null) { @@ -266,5 +264,4 @@ public void onComplete(final ClientResponse response) { } } } - } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java index eb3cfa145138b..abafda7af43f8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java @@ -33,21 +33,23 @@ public class RequestManagers { public final Optional coordinatorRequestManager; public final Optional commitRequestManager; public final ListOffsetsRequestManager listOffsetsRequestManager; + public final TopicMetadataRequestManager topicMetadataRequestManager; private final List> entries; public RequestManagers(ListOffsetsRequestManager listOffsetsRequestManager, + TopicMetadataRequestManager topicMetadataRequestManager, Optional coordinatorRequestManager, Optional 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> list = new ArrayList<>(); list.add(coordinatorRequestManager); list.add(commitRequestManager); list.add(Optional.of(listOffsetsRequestManager)); + list.add(Optional.of(topicMetadataRequestManager)); entries = Collections.unmodifiableList(list); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java new file mode 100644 index 0000000000000..5f95bd4b63c3e --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java @@ -0,0 +1,207 @@ +/* + * 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 org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.errors.InvalidTopicException; +import org.apache.kafka.common.errors.RetriableException; +import org.apache.kafka.common.errors.TopicAuthorizationException; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.utils.LogContext; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +/** + *

    + * Manages the state of the topic metadata requests. The manager returns the + * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when the request is ready to + * be sent. This manager specifically handles the following user API calls: + *

      + *
    • listOffsets
      • + *
    • partitionsFor
      • + *
    + *

    + *

    + * The manager checks the state of the {@link CompletableTopicMetadataRequest} before sending a new one, to + * prevent sending without backing off from the previous attempts. + * It also checks the state of the inflight requests to prevent overwhelming the broker with duplicated requests. + * The {@code inflightRequests} is memoized by the topic name. If all topics are requested, then we use {@code + * null} as the key. Once the request is completed successfully, the entry is removed. + *

    + */ +public class TopicMetadataRequestManager implements RequestManager { + private final boolean allowAutoTopicCreation; + private final Map inflightRequests; + private final long retryBackoffMs; + private final Logger log; + private final LogContext logContext; + + public TopicMetadataRequestManager(final LogContext logContext, final ConsumerConfig config) { + this.logContext = logContext; + this.log = logContext.logger(this.getClass()); + this.inflightRequests = new HashMap<>(); + this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG); + this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG); + } + + @Override + public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { + List requests = inflightRequests.values().stream() + .map(req -> req.send(currentTimeMs)) + .filter(Optional::isPresent) + .map(Optional::get) + .collect(Collectors.toList()); + return requests.isEmpty() ? + new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>()) : + new NetworkClientDelegate.PollResult(0, Collections.unmodifiableList(requests)); + } + + /** + * return the future of the metadata request. If the same topic was requested but incompleted, return the same + * future. + * + * @param topic to be requested. If empty, return the metadata for all topics. + * @return the future of the metadata request. + */ + public CompletableFuture>> requestTopicMetadata(final Optional topic) { + String topicName = topic.orElse(null); + if (inflightRequests.containsKey(topicName)) { + return inflightRequests.get(topicName).future(); + } + + CompletableTopicMetadataRequest newRequest = new CompletableTopicMetadataRequest( + logContext, + topic, + retryBackoffMs); + inflightRequests.put(topicName, newRequest); + return newRequest.future(); + } + + // Visible for testing + List inflightRequests() { + return new ArrayList<>(inflightRequests.values()); + } + + class CompletableTopicMetadataRequest extends CompletableRequest>> { + private final Optional topic; + private final RequestState state; + + public CompletableTopicMetadataRequest(final LogContext logContext, + final Optional topic, + final long retryBackoffMs) { + this.topic = topic; + this.state = new RequestState(logContext, retryBackoffMs); + } + + /** + * prepare the metadata request and return an + * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed. + */ + private Optional send(final long currentTimeMs) { + if (!this.state.canSendRequest(currentTimeMs)) { + return Optional.empty(); + } + this.state.onSendAttempt(currentTimeMs); + + final MetadataRequest.Builder request; + if (topic.isPresent()) { + request = new MetadataRequest.Builder(Collections.singletonList(topic.get()), allowAutoTopicCreation); + } else { + request = MetadataRequest.Builder.allTopics(); + } + + final NetworkClientDelegate.UnsentRequest unsent = new NetworkClientDelegate.UnsentRequest( + request, + Optional.empty(), + (response, exception) -> { + if (exception != null) { + this.future().completeExceptionally(new KafkaException(exception)); + inflightRequests.remove(topic.orElse(null)); + return; + } + + try { + Map> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody()); + future().complete(res); + inflightRequests.remove(topic.orElse(null)); + } catch (RetriableException e) { + this.state.onFailedAttempt(currentTimeMs); + } catch (Exception t) { + this.future().completeExceptionally(t); + inflightRequests.remove(topic.orElse(null)); + } + }); + return Optional.of(unsent); + } + + private Map> handleTopicMetadataResponse(final MetadataResponse response) { + Cluster cluster = response.buildCluster(); + + final Set unauthorizedTopics = cluster.unauthorizedTopics(); + if (!unauthorizedTopics.isEmpty()) + throw new TopicAuthorizationException(unauthorizedTopics); + + Map errors = response.errors(); + if (!errors.isEmpty()) { + // if there were errors, we need to check whether they were fatal or whether + // we should just retry + + log.debug("Topic metadata fetch included errors: {}", errors); + + for (Map.Entry errorEntry : errors.entrySet()) { + String topic = errorEntry.getKey(); + Errors error = errorEntry.getValue(); + + if (error == Errors.INVALID_TOPIC_EXCEPTION) + throw new InvalidTopicException("Topic '" + topic + "' is invalid"); + else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) + // if a requested topic is unknown, we just continue and let it be absent + // in the returned map + continue; + else if (error.exception() instanceof RetriableException) { + throw error.exception(); + } else + throw new KafkaException("Unexpected error fetching metadata for topic " + topic, + error.exception()); + } + } + + HashMap> topicsPartitionInfos = new HashMap<>(); + for (String topic : cluster.topics()) + topicsPartitionInfos.put(topic, cluster.partitionsForTopic(topic)); + return topicsPartitionInfos; + } + + public String topic() { + return topic.orElse(null); + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java index 1cad6e399ef36..1e8a10730e90d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java @@ -25,7 +25,7 @@ public abstract class ApplicationEvent { public enum Type { NOOP, COMMIT, POLL, FETCH, FETCH_COMMITTED_OFFSET, METADATA_UPDATE, UNSUBSCRIBE, - LIST_OFFSETS, + LIST_OFFSETS, TOPIC_METADATA } protected final Type type; diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java index 23619beaebc28..593e2762a74d1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java @@ -20,9 +20,14 @@ import org.apache.kafka.clients.consumer.internals.ConsumerMetadata; import org.apache.kafka.clients.consumer.internals.RequestManagers; import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.PartitionInfo; +import java.util.List; +import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; public class ApplicationEventProcessor { @@ -53,6 +58,8 @@ public boolean process(final ApplicationEvent event) { return process((OffsetFetchApplicationEvent) event); case METADATA_UPDATE: return process((MetadataUpdateApplicationEvent) event); + case TOPIC_METADATA: + return process((TopicMetadataApplicationEvent) event); case UNSUBSCRIBE: return process((UnsubscribeApplicationEvent) event); case LIST_OFFSETS: @@ -92,13 +99,7 @@ private boolean process(final CommitApplicationEvent event) { } CommitRequestManager manager = requestManagers.commitRequestManager.get(); - manager.addOffsetCommitRequest(event.offsets()).whenComplete((r, e) -> { - if (e != null) { - event.future().completeExceptionally(e); - return; - } - event.future().complete(null); - }); + event.chain(manager.addOffsetCommitRequest(event.offsets())); return true; } @@ -128,14 +129,21 @@ private boolean process(final UnsubscribeApplicationEvent event) { private boolean process(final ListOffsetsApplicationEvent event) { requestManagers.listOffsetsRequestManager.fetchOffsets(event.partitions, event.timestamp, - event.requireTimestamps) - .whenComplete((result, error) -> { - if (error != null) { - event.future().completeExceptionally(error); - return; - } - event.future().complete(result); - }); + event.requireTimestamps) + .whenComplete((result, error) -> { + if (error != null) { + event.future().completeExceptionally(error); + return; + } + event.future().complete(result); + }); + return true; + } + + private boolean process(final TopicMetadataApplicationEvent event) { + final CompletableFuture>> future = + this.requestManagers.topicMetadataRequestManager.requestTopicMetadata(Optional.of(event.topic())); + event.chain(future); return true; } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java index 3f4a9b8db11a4..eb411ca818e76 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java @@ -23,7 +23,6 @@ import java.util.Map; public class CommitApplicationEvent extends CompletableApplicationEvent { - private final Map offsets; public CommitApplicationEvent(final Map offsets) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java index 5a577906d9807..a286429a28d0e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java @@ -20,6 +20,7 @@ import org.apache.kafka.common.errors.InterruptException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.utils.Timer; + import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -54,4 +55,13 @@ public T get(Timer timer) { } } + public void chain(final CompletableFuture providedFuture) { + providedFuture.whenComplete((value, exception) -> { + if (exception != null) { + this.future.completeExceptionally(exception); + } else { + this.future.complete(value); + } + }); + } } \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicMetadataApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicMetadataApplicationEvent.java new file mode 100644 index 0000000000000..aab1bb71db26a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicMetadataApplicationEvent.java @@ -0,0 +1,34 @@ +/* + * 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.common.PartitionInfo; + +import java.util.List; +import java.util.Map; + +public class TopicMetadataApplicationEvent extends CompletableApplicationEvent>> { + private final String topic; + public TopicMetadataApplicationEvent(final String topic) { + super(Type.TOPIC_METADATA); + this.topic = topic; + } + + public String topic() { + return topic; + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java index 7bfcd6916bec1..691c58165b29f 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -120,7 +120,7 @@ public void testPoll_EnsureCorrectInflightRequestBufferSize() { offsets2.put(new TopicPartition("test", 4), new OffsetAndMetadata(20L)); // Add the requests to the CommitRequestManager and store their futures - ArrayList> commitFutures = new ArrayList<>(); + ArrayList> commitFutures = new ArrayList<>(); ArrayList>> fetchFutures = new ArrayList<>(); commitFutures.add(commitManager.addOffsetCommitRequest(offsets1)); fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 0)))); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java index 2b2bd75ceeb25..3cacf96853af2 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java @@ -24,6 +24,7 @@ import org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.MetadataUpdateApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.NoopApplicationEvent; +import org.apache.kafka.clients.consumer.internals.events.TopicMetadataApplicationEvent; import org.apache.kafka.common.message.FindCoordinatorRequestData; import org.apache.kafka.common.requests.FindCoordinatorRequest; import org.apache.kafka.common.serialization.StringDeserializer; @@ -41,6 +42,7 @@ import java.util.Optional; import java.util.Properties; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingQueue; import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; @@ -51,6 +53,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -72,6 +75,7 @@ public class DefaultBackgroundThreadTest { private int requestTimeoutMs = 500; private GroupState groupState; private CommitRequestManager commitManager; + private TopicMetadataRequestManager topicMetadataRequestManager; @BeforeEach @SuppressWarnings("unchecked") @@ -96,6 +100,7 @@ public void setup() { true); this.groupState = new GroupState(rebalanceConfig); this.commitManager = mock(CommitRequestManager.class); + this.topicMetadataRequestManager = mock(TopicMetadataRequestManager.class); } @Test @@ -171,6 +176,17 @@ void testFindCoordinator() { backgroundThread.close(); } + @Test + void testFetchTopicMetadata() { + this.applicationEventsQueue = new LinkedBlockingQueue<>(); + DefaultBackgroundThread backgroundThread = mockBackgroundThread(); + when(this.topicMetadataRequestManager.requestTopicMetadata(Optional.of(anyString()))).thenReturn(new CompletableFuture<>()); + this.applicationEventsQueue.add(new TopicMetadataApplicationEvent("topic")); + backgroundThread.runOnce(); + verify(applicationEventProcessor).process(any(TopicMetadataApplicationEvent.class)); + backgroundThread.close(); + } + @Test void testPollResultTimer() { DefaultBackgroundThread backgroundThread = mockBackgroundThread(); @@ -188,9 +204,10 @@ void testPollResultTimer() { private RequestManagers mockRequestManagers() { return new RequestManagers( - listOffsetsRequestManager, - Optional.of(coordinatorManager), - Optional.of(commitManager)); + listOffsetsRequestManager, + topicMetadataRequestManager, + Optional.of(coordinatorManager), + Optional.of(commitManager)); } private static NetworkClientDelegate.UnsentRequest findCoordinatorUnsentRequest( @@ -213,30 +230,32 @@ private DefaultBackgroundThread mockBackgroundThread() { properties.put(RETRY_BACKOFF_MS_CONFIG, RETRY_BACKOFF_MS); return new DefaultBackgroundThread(this.time, - new ConsumerConfig(properties), - new LogContext(), - applicationEventsQueue, - backgroundEventsQueue, - this.metadata, - this.networkClient, - this.subscriptionState, - this.groupState, - this.errorEventHandler, - applicationEventProcessor, - this.coordinatorManager, - this.commitManager, - this.listOffsetsRequestManager); + new ConsumerConfig(properties), + new LogContext(), + applicationEventsQueue, + backgroundEventsQueue, + this.metadata, + this.networkClient, + this.subscriptionState, + this.groupState, + this.errorEventHandler, + applicationEventProcessor, + this.coordinatorManager, + this.commitManager, + this.listOffsetsRequestManager, + this.topicMetadataRequestManager); + } private NetworkClientDelegate.PollResult mockPollCoordinatorResult() { return new NetworkClientDelegate.PollResult( - RETRY_BACKOFF_MS, - Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); + RETRY_BACKOFF_MS, + Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); } private NetworkClientDelegate.PollResult mockPollCommitResult() { return new NetworkClientDelegate.PollResult( - RETRY_BACKOFF_MS, - Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); + RETRY_BACKOFF_MS, + Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/IdempotentCloserTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/IdempotentCloserTest.java index 0e892f55d5fc3..69483fa223c18 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/IdempotentCloserTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/IdempotentCloserTest.java @@ -28,7 +28,7 @@ public class IdempotentCloserTest { - private static final Runnable CALLBACK_NO_OP = () -> {}; + private static final Runnable CALLBACK_NO_OP = () -> { }; private static final Runnable CALLBACK_WITH_RUNTIME_EXCEPTION = () -> { throw new RuntimeException("Simulated error during callback"); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManagerTest.java new file mode 100644 index 0000000000000..7e9c608f75a80 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManagerTest.java @@ -0,0 +1,191 @@ +/* + * 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 org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.apache.kafka.common.requests.RequestTestUtils; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TopicMetadataRequestManagerTest { + private MockTime time; + private TopicMetadataRequestManager topicMetadataRequestManager; + + private Properties props; + + @BeforeEach + public void setup() { + this.time = new MockTime(); + this.props = new Properties(); + this.props.put(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG, 100); + this.props.put(ALLOW_AUTO_CREATE_TOPICS_CONFIG, false); + this.props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + this.props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + this.topicMetadataRequestManager = new TopicMetadataRequestManager( + new LogContext(), + new ConsumerConfig(props)); + } + + @ParameterizedTest + @MethodSource("topicsProvider") + public void testPoll_SuccessfulRequestTopicMetadata(Optional topic) { + this.topicMetadataRequestManager.requestTopicMetadata(topic); + this.time.sleep(100); + NetworkClientDelegate.PollResult res = this.topicMetadataRequestManager.poll(this.time.milliseconds()); + assertEquals(1, res.unsentRequests.size()); + } + + @ParameterizedTest + @MethodSource("exceptionProvider") + public void testExceptionAndInflightRequests(final Errors error, final boolean shouldRetry) { + String topic = "hello"; + this.topicMetadataRequestManager.requestTopicMetadata(Optional.of("hello")); + this.time.sleep(100); + NetworkClientDelegate.PollResult res = this.topicMetadataRequestManager.poll(this.time.milliseconds()); + res.unsentRequests.get(0).future().complete(buildTopicMetadataClientResponse( + res.unsentRequests.get(0), + topic, + error)); + List inflights = this.topicMetadataRequestManager.inflightRequests(); + + if (shouldRetry) { + assertEquals(1, inflights.size()); + assertEquals(topic, inflights.get(0).topic()); + } else { + assertEquals(0, inflights.size()); + } + } + + @Test + public void testEnsureRequestRemovedFromInflightsOnErrorResponse() { + this.topicMetadataRequestManager.requestTopicMetadata(Optional.of("hello")); + this.time.sleep(100); + NetworkClientDelegate.PollResult res = this.topicMetadataRequestManager.poll(this.time.milliseconds()); + res.unsentRequests.get(0).future().completeExceptionally(new KafkaException("some error")); + + List inflights = this.topicMetadataRequestManager.inflightRequests(); + assertTrue(inflights.isEmpty()); + } + + @Test + public void testSendingTheSameRequest() { + final String topic = "hello"; + CompletableFuture>> future = this.topicMetadataRequestManager.requestTopicMetadata(Optional.of(topic)); + CompletableFuture>> future2 = + this.topicMetadataRequestManager.requestTopicMetadata(Optional.of(topic)); + this.time.sleep(100); + NetworkClientDelegate.PollResult res = this.topicMetadataRequestManager.poll(this.time.milliseconds()); + assertEquals(1, res.unsentRequests.size()); + + res.unsentRequests.get(0).future().complete(buildTopicMetadataClientResponse( + res.unsentRequests.get(0), + topic, + Errors.NONE)); + + assertTrue(future.isDone()); + assertFalse(future.isCompletedExceptionally()); + assertTrue(future2.isDone()); + assertFalse(future2.isCompletedExceptionally()); + } + + private ClientResponse buildTopicMetadataClientResponse( + final NetworkClientDelegate.UnsentRequest request, + final String topic, + final Errors error) { + AbstractRequest abstractRequest = request.requestBuilder().build(); + assertTrue(abstractRequest instanceof MetadataRequest); + MetadataRequest metadataRequest = (MetadataRequest) abstractRequest; + Cluster cluster = mockCluster(3, 0); + List topics = new ArrayList<>(); + topics.add(new MetadataResponse.TopicMetadata(error, topic, false, + Collections.emptyList())); + final MetadataResponse metadataResponse = RequestTestUtils.metadataResponse(cluster.nodes(), + cluster.clusterResource().clusterId(), + cluster.controller().id(), + topics); + return new ClientResponse( + new RequestHeader(ApiKeys.METADATA, metadataRequest.version(), "mockClientId", 1), + request.callback(), + "-1", + time.milliseconds(), + time.milliseconds(), + false, + null, + null, + metadataResponse); + } + + private static Cluster mockCluster(final int numNodes, final int controllerIndex) { + HashMap nodes = new HashMap<>(); + for (int i = 0; i < numNodes; i++) + nodes.put(i, new Node(i, "localhost", 8121 + i)); + return new Cluster("mockClusterId", nodes.values(), + Collections.emptySet(), Collections.emptySet(), + Collections.emptySet(), nodes.get(controllerIndex)); + } + + + private static Collection topicsProvider() { + return Arrays.asList( + Arguments.of(Optional.of("topic1")), + Arguments.of(Optional.empty())); + } + + private static Collection exceptionProvider() { + return Arrays.asList( + Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, false), + Arguments.of(Errors.INVALID_TOPIC_EXCEPTION, false), + Arguments.of(Errors.UNKNOWN_SERVER_ERROR, false), + Arguments.of(Errors.NETWORK_EXCEPTION, true), + Arguments.of(Errors.NONE, false)); + } +}