You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "philipnee (via GitHub)" <gi...@apache.org> on 2023/09/13 19:56:50 UTC
[GitHub] [kafka] philipnee opened a new pull request, #14386: KAFKA-14960: TopicMetadata request manager
philipnee opened a new pull request, #14386:
URL: https://github.com/apache/kafka/pull/14386
TopicMetadataRequestManager is responsible for sending topic metadata requests. The manager manages API requests and build the request accordingly. All topic metadata requests are chained, if requesting the same topic, to avoid sending requests repeatedly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326800659
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ if (e instanceof TimeoutException) {
Review Comment:
Hey @junrao - Sorry for writing such a long reply - I really need to brush up my memory about the network module as well as the TopicMetadataRequest...
In the [current code](https://github.com/apache/kafka/blob/b49013b73efa25466652d8d8122974e60c927ec4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java#L99-L149) I think the following is how responses should be handled.
For the case of hard failure that the callback receives a non-null exception:
We should remove the inflight request for all non-retriable and TimeoutException. We should continue to retry for RetriableException until timer runs out.
For the case of soft failure (the server respond with an error code), it seems like the right thing to do is to continue to retry except for the non-retriable exception.
It is worth noting that, if the request timeout (API level timeout), I think the callback is completed with TimeoutException hard failure, which should cause the exception being purged from the inflight request.
To your question: if the responses include a retriable exception, I believe the right thing to do is to continue to retry until timeout.
In fact - it seems like `if (e instanceof TimeoutException)` (that I added in the last commit) is actually incorrect. Because it is a RetriableException at it should be retried per:
```
if (error.exception() instanceof RetriableException)
shouldRetry = true;
```
It seems like what's missing here is handling the hard RetriableException and TimeoutException. I will fix the code and add test cases to cover the scenarios above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325349813
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -106,6 +104,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList());
}
+ pendingRequests.inflightOffsetFetches.forEach(System.out::println);
Review Comment:
🤦
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -106,6 +104,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList());
}
+ pendingRequests.inflightOffsetFetches.forEach(System.out::println);
Review Comment:
🤦
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327735512
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ private final Logger log;
+ private final LogContext logContext;
+
+ public TopicMetadataRequestManager(final LogContext logContext, final ConsumerConfig config) {
+ this.logContext = logContext;
Review Comment:
sorry - got into the habit of using `this`. 😅
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ private final Logger log;
+ private final LogContext logContext;
+
+ public TopicMetadataRequestManager(final LogContext logContext, final ConsumerConfig config) {
+ this.logContext = logContext;
Review Comment:
sorry - got into the habit of using `this`. 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326449328
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManagerTest.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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;
Review Comment:
This is only used in the setup so no need to keep it at the class level
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kirktrue commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326361661
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
- if (exception.isPresent()) {
- throw new RuntimeException(exception.get());
+ this.offsets = Collections.unmodifiableMap(offsets);
+
+ for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Review Comment:
If your PR goes in before mine, you can make a comment and I'll add the `toString` changes into mine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14386:
URL: https://github.com/apache/kafka/pull/14386#issuecomment-1728753394
@junrao @lianetm - Thanks for another round of reviews. The latest build passed with the following test failures
```
Build / JDK 17 and Scala 2.13 / testBecomeFollowerWhileOldClientFetchInPurgatory() – kafka.server.ReplicaManagerTest
<1s
Build / JDK 17 and Scala 2.13 / [1] enableRemoteStorage=true – kafka.server.ReplicaManagerTest
<1s
Build / JDK 20 and Scala 2.13 / testRackAwareRangeAssignor() – integration.kafka.server.FetchFromFollowerIntegrationTest
46s
Build / JDK 20 and Scala 2.13 / testBumpTransactionalEpoch(String).quorum=zk – kafka.api.TransactionsTest
13s
Build / JDK 20 and Scala 2.13 / testCommitTransactionTimeout(String).quorum=kraft – kafka.api.TransactionsTest
9s
Build / JDK 20 and Scala 2.13 / executeTieredStorageTest(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.DeleteTopicTest
41s
Build / JDK 11 and Scala 2.13 / testCloseDuringRebalance() – kafka.api.ConsumerBounceTest
10s
Build / JDK 11 and Scala 2.13 / [2] Type=Raft-Isolated, Name=testDescribeQuorumReplicationSuccessful, MetadataVersion=3.7-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
1m 8s
Build / JDK 8 and Scala 2.12 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 31s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
```
I believe they are irrelevant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327799007
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
+ }
+
+ private void handleException(final Throwable exception, final long responseTimeMs) {
+ if (exception instanceof TimeoutException || !(exception instanceof RetriableException)) {
Review Comment:
Just careful here because the current `TopicMetadataFetcher` handles 2 timeout cases differently regarding retries:
- TimeoutException contained in a MetadataResponse as a response error is retried [here](https://github.com/apache/kafka/blob/f46db86d34f9e5fe1b0d7604306a5108a89c113e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java#L133)
- TimeoutException in getting the actual MetadataResponse throws exception [here](https://github.com/apache/kafka/blob/f46db86d34f9e5fe1b0d7604306a5108a89c113e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java#L151)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1330555404
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -451,11 +443,11 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetch
/**
* Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the
* {@code inflightOffsetFetches} to bookkeep all of the inflight requests.
- *
+ * <p>
Review Comment:
was the intention here to add an extra line in-between? should be `<p/>` if so
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325355774
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
Review Comment:
As well as listTopics() - this is addressed in a separate PR: Would you be ok if we split this into multiple PRs? If you think we should remove the documentation due to inaccurate information, I'm willing to do so.
The PR's commit is c71a18c95937dd18171f60afb4fd263ea39e9a1b - I anticipate @kirktrue will pick this into trunk soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325352296
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -381,41 +377,41 @@ public String toString() {
}
/**
- * <p>This is used to stage the unsent {@link OffsetCommitRequestState} and {@link OffsetFetchRequestState}.
+ * <p>This is used to stage the unsent {@link OffsetCommitRequest} and {@link OffsetFetchRequestState}.
* <li>unsentOffsetCommits holds the offset commit requests that have not been sent out</>
* <li>unsentOffsetFetches holds the offset fetch requests that have not been sent out</li>
- * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but incompleted</>.
- *
+ * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but not completed</>.
+ * <p>
* {@code addOffsetFetchRequest} dedupes the requests to avoid sending the same requests.
*/
class PendingRequests {
// Queue is used to ensure the sequence of commit
- Queue<OffsetCommitRequestState> unsentOffsetCommits = new LinkedList<>();
+ Queue<OffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>();
List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>();
- public boolean hasUnsentRequests() {
+ boolean hasUnsentRequests() {
Review Comment:
Sorry - leaving this protected as it is used in one of the test to verify there's no more unsent request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #14386:
URL: https://github.com/apache/kafka/pull/14386
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1330551023
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -395,27 +386,28 @@ class PendingRequests {
List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>();
List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>();
- public boolean hasUnsentRequests() {
+ // Visible for teseting
+ boolean hasUnsentRequests() {
return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
}
- public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ CompletableFuture<Void> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
// TODO: Dedupe committing the same offsets to the same partitions
OffsetCommitRequestState request = new OffsetCommitRequestState(
offsets,
groupState.groupId,
groupState.groupInstanceId.orElse(null),
groupState.generation);
unsentOffsetCommits.add(request);
- return request.future();
+ return request.future;
}
/**
- * <p>Adding an offset fetch request to the outgoing buffer. If the same request was made, we chain the future
- * to the existing one.
+ * <p>Adding an offset fetch request to the outgoing buffer. If the same request was made, we chain the future
+ * to the existing one.
*
- * <p>If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches}
- * upon completion.</>
+ * <p>If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
Review Comment:
nit: wrong tag </> and other unclosed ones above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14386:
URL: https://github.com/apache/kafka/pull/14386#issuecomment-1724712594
@junrao @lianetm - I hope I've addressed most, if not all, comments that you pointed out. So, thank you both for reviewing my code.
We did have an offline discussion about user API timeout handling, and a bug report is filed here [KAFKA-15475](https://issues.apache.org/jira/browse/KAFKA-15475). I will submit a patch after this PR to address this small gap.
Lastly, here are the failing tests:
```
Build / JDK 20 and Scala 2.13 / testReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
53s
Build / JDK 17 and Scala 2.13 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 24s
Build / JDK 17 and Scala 2.13 / testCreateTokenForOtherUserFails(String).quorum=kraft – kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest
9s
Build / JDK 11 and Scala 2.13 / testConfigurationOperations() – org.apache.kafka.controller.QuorumControllerTest
<1s
Build / JDK 11 and Scala 2.13 / executeTieredStorageTest(String).quorum=zk – org.apache.kafka.tiered.storage.integration.EnableRemoteLogOnTopicTest
43s
Build / JDK 8 and Scala 2.12 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 25s
```
I believe these are flaky.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326800659
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ if (e instanceof TimeoutException) {
Review Comment:
Hey @junrao - Sorry for writing such a long reply - I really need to brush up my memory about the network module as well as the TopicMetadataRequest...
In the [current code](https://github.com/apache/kafka/blob/b49013b73efa25466652d8d8122974e60c927ec4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java#L99-L149) I think the following is how responses should be handled.
For the case of hard failure that the callback receives a non-null exception:
We should remove the inflight request for all non-retriable and TimeoutException because we need to fail the user requests and topic the retry. However, we should continue to retry for RetriableException until timer runs out.
For the case of soft failure (the server respond with an error code), it seems like the right thing to do is to continue to retry except for the non-retriable exception.
It is worth noting that: If the request timed-out (API level timeout), callback is completed with TimeoutException hard failure (in the network delegate module), which should eventually causes the exception to be purged from the inflight request as mentioned above.
To your question: if the responses include a retriable exception, I believe the right thing to do is to continue to retry until the timer runs out.
In fact - it seems like `if (e instanceof TimeoutException)` (that I added in the last commit) is actually incorrect. Because it is a RetriableException at it should be retried per:
```
if (error.exception() instanceof RetriableException)
shouldRetry = true;
```
What's missing here is handling the hard RetriableException and TimeoutException. I will fix the code and add test cases to cover the scenarios above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327733352
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
Review Comment:
Hi Jun - understood making repeated system calls is bad. However - we do need to get the time when the response is received, so this is kind of the lazy way to do it.
I think it is possible to expose the time during the callback/future completion during the network poll (because it is computed at the top of the background thread loop and pass down to network poll.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1330455761
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ private final Logger log;
+ private final LogContext logContext;
+
+ public TopicMetadataRequestManager(final LogContext context, final ConsumerConfig config) {
+ logContext = context;
+ log = logContext.logger(this.getClass());
Review Comment:
Could we remove `this`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -91,7 +90,7 @@ public CommitRequestManager(
}
/**
- * Poll for the {@link OffsetFetchRequest} and {@link OffsetCommitRequest} request if there's any. The function will
+ * Poll for the {@link OffsetFetchRequest} and {@link OffsetCommitRequestState} request if there's any. The function will
Review Comment:
Hmm, OffsetCommitRequestState should be OffsetCommitRequest, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ private final Logger log;
+ private final LogContext logContext;
+
+ public TopicMetadataRequestManager(final LogContext context, final ConsumerConfig config) {
+ logContext = context;
+ log = logContext.logger(this.getClass());
+ inflightRequests = new HashMap<>();
+ retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
+ retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request =
+ topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ return Optional.of(createUnsentRequest(request));
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ if (exception == null) {
+ handleResponse(response, response.receivedTimeMs());
+ return;
+ }
+
+ if (exception instanceof RetriableException) {
+ // We continue to retry on RetriableException
+ // TODO: TimeoutException will continue to retry despite user API timeout.
+ onFailedAttempt(response.receivedTimeMs());
+ } else {
+ completeFutureAndRemoveRequest(new KafkaException(exception));
+ }
+ }
+
+ private void handleResponse(final ClientResponse response, final long responseTimeMs) {
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ onFailedAttempt(responseTimeMs);
+ } catch (Exception t) {
+ completeFutureAndRemoveRequest(t);
+ }
+ }
+
+ private void completeFutureAndRemoveRequest(final Throwable throwable) {
+ this.future.completeExceptionally(throwable);
Review Comment:
Could we remove `this`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -395,27 +386,28 @@ class PendingRequests {
List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>();
List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>();
- public boolean hasUnsentRequests() {
+ // Visible for teseting
Review Comment:
typo teseting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1329213351
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -298,8 +289,8 @@ private void onFailure(final long currentTimeMs,
}
private void retry(final long currentTimeMs) {
- onFailedAttempt(currentTimeMs);
- onSendAttempt(currentTimeMs);
+ this.onFailedAttempt(currentTimeMs);
Review Comment:
Addressed, see line #259
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325353270
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -203,16 +204,17 @@ public static class UnsentRequest {
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) {
Review Comment:
I think it is according to my IDE. I think it is used in CommitRequestManager (line 219 and 256) and TopicMetadataRequestManger (line 145)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325364414
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ this.onFailedAttempt(currentTimeMs);
Review Comment:
Hi Jun - Thanks for the catch - I added a test case in the TopicMetadataRequestManagerTest:146
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14386:
URL: https://github.com/apache/kafka/pull/14386#issuecomment-1718779561
Hey @junrao - Addressed your comments. I'm holding off on your request for `Could we change toString to include the fields in the parent class?` as I think it would be easier if I do that consistently for all other CompletableApplicationEvents. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326612426
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
Review Comment:
Yes, this is fine.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManagerTest.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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<String> 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),
+ Optional.of(topic),
+ error));
+ List<TopicMetadataRequestManager.TopicMetadataRequestState> inflights = this.topicMetadataRequestManager.inflightRequests();
+
+ if (shouldRetry) {
+ assertEquals(1, inflights.size());
+ assertEquals(topic, inflights.get(0).topic().orElse(null));
+ } 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<TopicMetadataRequestManager.TopicMetadataRequestState> inflights = this.topicMetadataRequestManager.inflightRequests();
+ assertTrue(inflights.isEmpty());
+ }
+
+ @ParameterizedTest
+ @MethodSource("topicsProvider")
+ public void testSendingTheSameRequest(Optional<String> topic) {
+ CompletableFuture<Map<String, List<PartitionInfo>>> future = this.topicMetadataRequestManager.requestTopicMetadata(topic);
+ CompletableFuture<Map<String, List<PartitionInfo>>> future2 =
+ this.topicMetadataRequestManager.requestTopicMetadata(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());
+ try {
+ future.get();
+ } catch (Throwable e) {
+ System.out.println(e.getMessage());
Review Comment:
In general, we should use logging instead of print.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java:
##########
@@ -287,14 +298,13 @@ public void testAssignmentChangeEvent() {
when(this.coordinatorManager.poll(anyLong())).thenReturn(mockPollCoordinatorResult());
when(this.commitManager.poll(anyLong())).thenReturn(mockPollCommitResult());
when(this.offsetsRequestManager.poll(anyLong())).thenReturn(emptyPollOffsetsRequestResult());
+ when(this.topicMetadataRequestManager.poll(anyLong())).thenReturn(emptyPollOffsetsRequestResult());
backgroundThread.runOnce();
verify(applicationEventProcessor).process(any(AssignmentChangeApplicationEvent.class));
verify(networkClient, times(1)).poll(anyLong(), anyLong());
verify(commitManager, times(1)).updateAutoCommitTimer(currentTimeMs);
verify(commitManager, times(1)).maybeAutoCommit(offset);
-
- backgroundThread.close();
Review Comment:
It seems that we close backgroundThread in other cases?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ if (e instanceof TimeoutException) {
Review Comment:
This captures the case when a single request takes long on the broker and times out. However, a separate case is that individual request returns on time but with other RetriableException. In that case, TopicMetadataRequestManager will not clean up inflightRequests even when the client request has timed out.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
- if (exception.isPresent()) {
- throw new RuntimeException(exception.get());
+ this.offsets = Collections.unmodifiableMap(offsets);
+
+ for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Review Comment:
Yes, it's fine to clean this up separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327372398
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -141,35 +142,57 @@ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTim
final MetadataRequest.Builder request;
request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
- .orElseGet(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);
- return;
- }
-
- try {
- Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
- future.complete(res);
- inflightRequests.remove(topic);
- } catch (RetriableException e) {
- if (e instanceof TimeoutException) {
- inflightRequests.remove(topic);
- }
- this.onFailedAttempt(currentTimeMs);
- } catch (Exception t) {
- this.future.completeExceptionally(t);
- inflightRequests.remove(topic);
- }
- });
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
return Optional.of(unsent);
}
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
Review Comment:
It seems that the exception handling logic here ends up in 2 places. Apart from the initial `handleException`, this `handleResponse` could be as well end up handling exceptions, in a very similar way (catch block on `handleResponse` vs the `handleException`). If we prefer to keep them separated because of the "`retryOnTimeout`" part, which seems to be what makes them different, it would be helpful maybe to add a comment to describe that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327568639
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
Review Comment:
Could we pass in currentTime from the caller?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ private final Logger log;
+ private final LogContext logContext;
+
+ public TopicMetadataRequestManager(final LogContext logContext, final ConsumerConfig config) {
+ this.logContext = logContext;
Review Comment:
Could we just get rid of `this`? Ditto below.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
+ }
+
+ private void handleException(final Throwable exception, final long responseTimeMs) {
+ if (exception instanceof TimeoutException || !(exception instanceof RetriableException)) {
Review Comment:
TimeoutException is a RetriableException. Should we just treat it as other RetriableException? That seems to be what the old consumer does.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
Review Comment:
Could we just get rid of `this`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
+ }
+
+ private void handleException(final Throwable exception, final long responseTimeMs) {
+ if (exception instanceof TimeoutException || !(exception instanceof RetriableException)) {
+ completeFutureAndRemoveRequest(new KafkaException(exception));
+ } else {
+ // We continue to retry on RetriableException
+ onFailedAttempt(responseTimeMs);
+ }
+ }
+
+ private void handleResponse(final ClientResponse response, final long responseTimeMs) {
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ onFailedAttempt(responseTimeMs);
Review Comment:
The case that I am wondering is the following.
DefaultBackgroundThread is in a retry loop for an inflightRequest. PrototypeAsyncConsumer.partitionsFor eventually calls CompletableApplicationEvent.get(Timer timer) and times out. Do we still keep the request in inflightRequests at that point? Is that a concern?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
Review Comment:
It would be useful to be consistent in convention. If we want to add `this` in front of methods in the parent class, we want to do this consistently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325341358
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
Review Comment:
🤦 - listTopics() instead.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
Review Comment:
🤦 - listTopics() instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327744776
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -298,8 +289,8 @@ private void onFailure(final long currentTimeMs,
}
private void retry(final long currentTimeMs) {
- onFailedAttempt(currentTimeMs);
- onSendAttempt(currentTimeMs);
+ this.onFailedAttempt(currentTimeMs);
Review Comment:
currentTimeMs is when the request is sent, not when the response is receive. Could we pass along `ClientResponse.receivedTimeMs`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
Review Comment:
Actually, could we use `ClientResponse.receivedTimeMs`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325040397
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -91,7 +89,7 @@ public CommitRequestManager(
}
/**
- * Poll for the {@link OffsetFetchRequest} and {@link OffsetCommitRequest} request if there's any. The function will
+ * Poll for the {@link OffsetFetchRequest} and {@link org.apache.kafka.common.requests.OffsetCommitRequest} request if there's any. The function will
Review Comment:
Do we need to spell out the package name for OffsetCommitRequest? It doesn't seem this is done consistently.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
Review Comment:
It doesn't seem that we have wired `TopicMetadataRequestManager` to the `partitionsFor` call?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -381,41 +377,41 @@ public String toString() {
}
/**
- * <p>This is used to stage the unsent {@link OffsetCommitRequestState} and {@link OffsetFetchRequestState}.
+ * <p>This is used to stage the unsent {@link OffsetCommitRequest} and {@link OffsetFetchRequestState}.
* <li>unsentOffsetCommits holds the offset commit requests that have not been sent out</>
* <li>unsentOffsetFetches holds the offset fetch requests that have not been sent out</li>
- * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but incompleted</>.
- *
+ * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but not completed</>.
+ * <p>
* {@code addOffsetFetchRequest} dedupes the requests to avoid sending the same requests.
*/
class PendingRequests {
// Queue is used to ensure the sequence of commit
- Queue<OffsetCommitRequestState> unsentOffsetCommits = new LinkedList<>();
+ Queue<OffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>();
List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>();
- public boolean hasUnsentRequests() {
+ boolean hasUnsentRequests() {
Review Comment:
Could this be private? Ditto below.
##########
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<Map<String, List<PartitionInfo>>> {
+ private final String topic;
+ public TopicMetadataApplicationEvent(final String topic) {
+ super(Type.TOPIC_METADATA);
+ this.topic = topic;
+ }
+
+ public String topic() {
+ return topic;
+ }
Review Comment:
Do we need to add `toString`, `equals` and `hashcode`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
Review Comment:
Is this true? `OffsetsRequestManager` says it handles listOffsets.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -220,7 +214,7 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
requestTopicDataMap.put(topicPartition.topic(), topic);
}
- OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder(
+ org.apache.kafka.common.requests.OffsetCommitRequest.Builder builder = new org.apache.kafka.common.requests.OffsetCommitRequest.Builder(
Review Comment:
Could we import the package?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
Review Comment:
memoized => memorized
null => Empty?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
Review Comment:
Do we need to spell out the package name?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ this.onFailedAttempt(currentTimeMs);
Review Comment:
This may be fine. But if a pending request is in a retry loop, but is already timed out, do we need logic to remove it from inflightRequests preemptively?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
Review Comment:
`isValid` is no longer used.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
Review Comment:
RequestState only allows one outstanding request at a time. Does the old consumer do the same?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -203,16 +204,17 @@ public static class UnsentRequest {
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) {
Review Comment:
This constructor seems unused?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -106,6 +104,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList());
}
+ pendingRequests.inflightOffsetFetches.forEach(System.out::println);
Review Comment:
Is this intended?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
- if (exception.isPresent()) {
- throw new RuntimeException(exception.get());
+ this.offsets = Collections.unmodifiableMap(offsets);
+
+ for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Review Comment:
This is an existing issue. Could we change `toString` to include the fields in the parent class? Do we need to add `equals` and `hashcode` for consistency?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -230,15 +224,20 @@ 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 {
public final Set<TopicPartition> requestedPartitions;
public final GroupState.Generation requestedGeneration;
- public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
-
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
Review Comment:
Could this be private?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325345190
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
Review Comment:
RequestState is a new class introduced in this consumer to handle exponential backoff. For managers that only send a single request (FindCoordinator, Heartbeat for example), the implementation is quite simple - if the current request failed, it will backoff and retry.
For managers handling multiple requests like this one and the commit request manager - the request state exists on per request basis. I'm sure there's a better way to do this, but the original go was to offer exponential backoff functionality.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326441512
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
Review Comment:
indentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1328231607
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -298,8 +289,8 @@ private void onFailure(final long currentTimeMs,
}
private void retry(final long currentTimeMs) {
- onFailedAttempt(currentTimeMs);
- onSendAttempt(currentTimeMs);
+ this.onFailedAttempt(currentTimeMs);
Review Comment:
Thanks for catching this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1329455102
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
+ }
+
+ private void handleException(final Throwable exception, final long responseTimeMs) {
+ if (exception instanceof TimeoutException || !(exception instanceof RetriableException)) {
Review Comment:
I think for now - the manager will continue to retry all RetriableException including TimeoutException. Currently, we don't have a way to cancel the request when the user API request times-out, so this is a gap there, but a jira is filed here: https://issues.apache.org/jira/browse/KAFKA-15475
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1330555404
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -451,11 +443,11 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetch
/**
* Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the
* {@code inflightOffsetFetches} to bookkeep all of the inflight requests.
- *
+ * <p>
Review Comment:
was the intention here to add an extra line in-between? should be <p/> if so
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325175686
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -381,41 +377,41 @@ public String toString() {
}
/**
- * <p>This is used to stage the unsent {@link OffsetCommitRequestState} and {@link OffsetFetchRequestState}.
+ * <p>This is used to stage the unsent {@link OffsetCommitRequest} and {@link OffsetFetchRequestState}.
* <li>unsentOffsetCommits holds the offset commit requests that have not been sent out</>
* <li>unsentOffsetFetches holds the offset fetch requests that have not been sent out</li>
- * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but incompleted</>.
- *
+ * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but not completed</>.
+ * <p>
* {@code addOffsetFetchRequest} dedupes the requests to avoid sending the same requests.
*/
class PendingRequests {
// Queue is used to ensure the sequence of commit
- Queue<OffsetCommitRequestState> unsentOffsetCommits = new LinkedList<>();
+ Queue<OffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>();
List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>();
- public boolean hasUnsentRequests() {
+ boolean hasUnsentRequests() {
Review Comment:
Sorry - I left out commit. Fixing those obvious mistakes right away.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1326800659
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(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);
+ return;
+ }
+
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ if (e instanceof TimeoutException) {
Review Comment:
Hey @junrao - Sorry for writing such a long reply - I really need to brush up my memory about the network module as well as the TopicMetadataRequest...
In the [current code](https://github.com/apache/kafka/blob/b49013b73efa25466652d8d8122974e60c927ec4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java#L99-L149) I think the following is how responses should be handled.
For the case of hard failure that the callback receives a non-null exception:
We should remove the inflight request for all non-retriable and TimeoutException because we need to fail the user requests and topic the retry. However, we should continue to retry for RetriableException until timer runs out.
For the case of soft failure (the server respond with an error code), it seems like the right thing to do is to continue to retry except for the non-retriable exception.
It is worth noting that, if the request timeout (API level timeout), I think the callback is completed with TimeoutException hard failure, which should cause the exception being purged from the inflight request.
To your question: if the responses include a retriable exception, I believe the right thing to do is to continue to retry until timeout.
In fact - it seems like `if (e instanceof TimeoutException)` (that I added in the last commit) is actually incorrect. Because it is a RetriableException at it should be retried per:
```
if (error.exception() instanceof RetriableException)
shouldRetry = true;
```
It seems like what's missing here is handling the hard RetriableException and TimeoutException. I will fix the code and add test cases to cover the scenarios above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327781850
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
+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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listTopics</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memorized by topic name. If all topics are requested, then we use {@code Optional
+ * .empty()} as the key.
+ * Once a request is completed successfully, its corresponding entry is removed.
+ * </p>
+ */
+
+public class TopicMetadataRequestManager implements RequestManager {
+ private final boolean allowAutoTopicCreation;
+ private final Map<Optional<String>, TopicMetadataRequestState> inflightRequests;
+ private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
+ 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.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ this.allowAutoTopicCreation = config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> 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. Return the existing future if a request for the same topic is already
+ * inflight.
+ *
+ * @param topic to be requested. If empty, return the metadata for all topics.
+ * @return the future of the metadata request.
+ */
+ public CompletableFuture<Map<String, List<PartitionInfo>>> requestTopicMetadata(final Optional<String> topic) {
+ if (inflightRequests.containsKey(topic)) {
+ return inflightRequests.get(topic).future;
+ }
+
+ TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
+ logContext,
+ topic,
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ inflightRequests.put(topic, newRequest);
+ return newRequest.future;
+ }
+
+ // Visible for testing
+ List<TopicMetadataRequestState> inflightRequests() {
+ return new ArrayList<>(inflightRequests.values());
+ }
+
+ class TopicMetadataRequestState extends RequestState {
+ private final Optional<String> topic;
+ CompletableFuture<Map<String, List<PartitionInfo>>> future;
+
+ public TopicMetadataRequestState(final LogContext logContext,
+ final Optional<String> topic,
+ final long retryBackoffMs,
+ final long retryBackoffMaxMs) {
+ super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs);
+ this.future = new CompletableFuture<>();
+ this.topic = topic;
+ }
+
+ /**
+ * prepare the metadata request and return an
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
+ */
+ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTimeMs) {
+ if (!this.canSendRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
+ this.onSendAttempt(currentTimeMs);
+
+ final MetadataRequest.Builder request;
+ request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
+ return Optional.of(unsent);
+ }
+
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
+ }
+
+ private void handleException(final Throwable exception, final long responseTimeMs) {
+ if (exception instanceof TimeoutException || !(exception instanceof RetriableException)) {
+ completeFutureAndRemoveRequest(new KafkaException(exception));
+ } else {
+ // We continue to retry on RetriableException
+ onFailedAttempt(responseTimeMs);
+ }
+ }
+
+ private void handleResponse(final ClientResponse response, final long responseTimeMs) {
+ try {
+ Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
+ future.complete(res);
+ inflightRequests.remove(topic);
+ } catch (RetriableException e) {
+ onFailedAttempt(responseTimeMs);
Review Comment:
Thanks Jun for raising this point - I think what we should do is to complete the request with a non-retriable exception to prevent the request from retrying. Otherwise - the request manager will never remove the request from the inflightRequest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327524919
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -141,35 +142,57 @@ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTim
final MetadataRequest.Builder request;
request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
- .orElseGet(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);
- return;
- }
-
- try {
- Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
- future.complete(res);
- inflightRequests.remove(topic);
- } catch (RetriableException e) {
- if (e instanceof TimeoutException) {
- inflightRequests.remove(topic);
- }
- this.onFailedAttempt(currentTimeMs);
- } catch (Exception t) {
- this.future.completeExceptionally(t);
- inflightRequests.remove(topic);
- }
- });
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
return Optional.of(unsent);
}
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
Review Comment:
I think it would be clearer and better to maintain/troubleshoot if error handling is in one place. Maybe having a single `handleException(final Throwable exception, final long responseTimeMs, boolean retryOnTimeout)`, called in both cases:
```
if (exception != null) {
// Handle hard errors
handleException(exception, responseTimeMs, false);
...
}
...
catch (Exception e) {
// Handle soft errors
handleException(e, responseTimeMs, true);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325349677
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
- if (exception.isPresent()) {
- throw new RuntimeException(exception.get());
+ this.offsets = Collections.unmodifiableMap(offsets);
+
+ for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Review Comment:
There are quite a few events out there that need to be refactored if we do it this way. Would it be ok for you to skip this for now and I will post a MINOR patch to address after this is closed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1325342579
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+/**
+ * <p>
+ * Manages the state of topic metadata requests. This manager returns a
+ * {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} when a request is ready to
+ * be sent. Specifically, this manager handles the following user API calls:
+ * </p>
+ * <ul>
+ * <li>listOffsets</li>
+ * <li>partitionsFor</li>
+ * </ul>
+ * <p>
+ * The manager checks the state of the {@link TopicMetadataRequestState} before sending a new one to
+ * prevent sending it without backing off from previous attempts.
+ * It also checks the state of inflight requests to avoid overwhelming the broker with duplicate requests.
+ * The {@code inflightRequests} are memoized by topic name. If all topics are requested, then {@code null} is used as the key.
Review Comment:
thanks - apologize for neglecting to update the doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1329215296
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java:
##########
@@ -38,8 +38,8 @@
* Whether there is an existing coordinator.
* Whether there is an inflight request.
* Whether the backoff timer has expired.
- * The {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} contains either a wait timer
Review Comment:
just cleaning up long references
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327470102
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -141,35 +142,57 @@ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTim
final MetadataRequest.Builder request;
request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
- .orElseGet(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);
- return;
- }
-
- try {
- Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
- future.complete(res);
- inflightRequests.remove(topic);
- } catch (RetriableException e) {
- if (e instanceof TimeoutException) {
- inflightRequests.remove(topic);
- }
- this.onFailedAttempt(currentTimeMs);
- } catch (Exception t) {
- this.future.completeExceptionally(t);
- inflightRequests.remove(topic);
- }
- });
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
return Optional.of(unsent);
}
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
Review Comment:
thanks, @lianetm - Do you think it would be clearer to keep them in 1 place? The intention was to handle hard failures in handleException and soft failures (i.e. returning an error code) in handleResponse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327515651
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java:
##########
@@ -19,22 +19,23 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class CommitApplicationEvent extends ApplicationEvent {
- final private CompletableFuture<Void> future;
+public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
final private Map<TopicPartition, OffsetAndMetadata> offsets;
public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT);
- this.offsets = offsets;
- Optional<Exception> exception = isValid(offsets);
- if (exception.isPresent()) {
- throw new RuntimeException(exception.get());
+ this.offsets = Collections.unmodifiableMap(offsets);
+
+ for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Review Comment:
@junrao - per your request. Here's the patch to address your comment: https://github.com/apache/kafka/pull/14391
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1330938409
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -451,11 +443,11 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetch
/**
* Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the
* {@code inflightOffsetFetches} to bookkeep all of the inflight requests.
- *
+ * <p>
Review Comment:
😓 - must be the residual from some edits..
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -451,11 +443,11 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetch
/**
* Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the
* {@code inflightOffsetFetches} to bookkeep all of the inflight requests.
- *
+ * <p>
Review Comment:
😓 - must be the residual from some edits..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14386: KAFKA-14960: TopicMetadata request manager
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327372398
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -141,35 +142,57 @@ private Optional<NetworkClientDelegate.UnsentRequest> send(final long currentTim
final MetadataRequest.Builder request;
request = topic.map(t -> new MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
- .orElseGet(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);
- return;
- }
-
- try {
- Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
- future.complete(res);
- inflightRequests.remove(topic);
- } catch (RetriableException e) {
- if (e instanceof TimeoutException) {
- inflightRequests.remove(topic);
- }
- this.onFailedAttempt(currentTimeMs);
- } catch (Exception t) {
- this.future.completeExceptionally(t);
- inflightRequests.remove(topic);
- }
- });
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent = createUnsentRequest(request);
return Optional.of(unsent);
}
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
Review Comment:
It seems that the exception handling logic here ends up in 2 places. Apart from the initial `handleException`, this `handleResponse` could be as well end up handling exceptions, in a very similar way (catch block on `handleResponse` vs the `handleException`). If we prefer to keep them separated because of the Timeout, which seems to be the case that makes them different, it would be helpful maybe to add a comment to describe that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org