You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dimitarndimitrov (via GitHub)" <gi...@apache.org> on 2023/03/21 20:34:21 UTC
[GitHub] [kafka] dimitarndimitrov opened a new pull request, #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
dimitarndimitrov opened a new pull request, #13432:
URL: https://github.com/apache/kafka/pull/13432
Aside from consistency with the implementation of newer APIs this also provides more resilient retries (e.g. in case of TimeoutExceptions) and simplifies the code to some extent.
One notable detail is the addition of unsupported version handling in the failure callback in the `AdminApiDriver` and as a new callback in the `AdminApiHandler`. The handler callback is invoked in case an `UnsupportedVersionException` is caught by the driver and returns a flag indicating whether the exception could be handled (like the similar callback in `Call`).
- It also seemed prudent to signal whether the exception was caught in the fulfillment or the lookup stage (the latter case could then be delegated to the lookup strategy), but I don't know if there are actual use-cases for that.
The change has been tested manually and with a new `KafkaAdminClientTest`. The retry aspect also unexpectedly stalled a Connect `AdminClientUnitTestEnv` test due to retrying after a timeout where the test expected an immediate failure.
Also a JMH benchmark for the deleted equivalent of the fulfillment stage of the API has been deleted and not replaced as I wasn't too sure whether I can recreate it after the `AdminApiDriver` refactoring.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163919103
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
Review Comment:
I actually am not sure if this is ideal. If we have batches for which there are no `MAX_TIMESTAMP` requests, we don't want to incorrectly be indicating that in `ListOffsetsRequest.Builder`. Calculating on the fly will also allow us to avoid the mutable state.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1164285539
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,29 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
+ if (spec.scope instanceof FulfillmentScope) {
+ Map<K, Throwable> unrecoverableFailures =
+ handler.handleUnsupportedVersionException(
+ (UnsupportedVersionException) t,
+ spec.keys);
+ completeExceptionally(unrecoverableFailures);
+ } else {
+ Map<K, Throwable> unrecoverableLookupFailures =
+ handler.lookupStrategy().handleUnsupportedVersionException(
+ (UnsupportedVersionException) t,
+ spec.keys);
+ completeLookupExceptionally(unrecoverableLookupFailures);
+ Set<K> keysToUnmap = spec.keys.stream()
+ .filter(k -> !unrecoverableLookupFailures.containsKey(k))
+ .collect(Collectors.toSet());
+ retryLookup(keysToUnmap);
+ }
Review Comment:
@dajac , @dengziming Let me know if you prefer to simplify this for now. If we go that route, I think we have to do `completeLookupExceptionally` instead of `retryLookup` for all, but I might be missing something.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170024821
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> getOffsetFromSpec(offsetSpecsByPartition.get(key)) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys, boolean isFulfillmentStage
+ ) {
+ // An UnsupportedVersionException can be addressed only in the fulfillment stage...
+ if (!isFulfillmentStage) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
+ }
+ Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ boolean containsNonMaxTimestampSpec = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ if (getOffsetFromSpec(offsetSpec) == ListOffsetsRequest.MAX_TIMESTAMP) {
+ maxTimestampPartitions.put(topicPartition, exception);
+ } else {
+ containsNonMaxTimestampSpec = true;
+ }
+ }
+ // ...only if it's caused by a MAX_TIMESTAMP spec for some partition, and only if there is at least
+ // one other partition with a non-MAX_TIMESTAMP spec for which the fulfillment stage can be retried.
+ if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
Review Comment:
> we can replace it with `boolean containsNonMaxTimestampSpec = maxTimestampPartitions.size() != keys.size()` or other simpler expression, so `if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec)` can be changed to `if (maxTimestampPartitions.isEmpty() || maxTimestampPartitions.size()==keys.size())`.
Done.
> And I think we can remove `containsNonMaxTimestampSpec`, if `containsNonMaxTimestampSpec` is false, then the `keys.stream().collect(Collectors.toMap(k -> k, k -> exception))` and `maxTimestampPartitions` are the same.
Done.
I had some concerns about the non-obvious folding of the `containsNonMaxTimestampSpec == false` case, but I will try to explain that in the comment and we'll save an unnecessary map creation.
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1147319995
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java:
##########
@@ -70,6 +71,22 @@
*/
ApiResult<K, V> handleResponse(Node broker, Set<K> keys, AbstractResponse response);
+ /**
+ * Callback that is invoked when a request hits an UnsupportedVersionException. If the
+ * exception can be handled and the request retried, the appropriate changes to the request
+ * should be applied and that should be indicated via the returned boolean value. If the
+ * exception is thrown during the lookup stage the handler should delegate the handling to the
+ * lookup strategy.
+ *
+ * @return True if the exception can be handled; false otherwise.
+ */
+ default boolean handleUnsupportedVersionException(
Review Comment:
I forgot the details about the whole call chain, but I think we should decide which keys (if any) can retry with lower version and which keys can not, keys which are not retriable should be put into the result, and others are left out of the result which will be retried automatically, similar to what we do in `handleResponse`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ if (offsetQuery == ListOffsetsRequest.MAX_TIMESTAMP) {
+ if (skipMaxTimestampRequests) {
Review Comment:
Can we simplify this to `this.supportsMaxTimestamp = topicPartitionOffsets.values().stream()
.anyMatch(timestamp -> timestamp == ListOffsetsRequest.MAX_TIMESTAMP)`?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
Review Comment:
How about reuse `CollectionUtils.groupPartitionsByTopic` to avoid this for loop.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -261,19 +262,30 @@ public void onFailure(
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
} else {
+ boolean isFulfillmentStage = spec.scope instanceof FulfillmentScope;
+ if (handleUnsupportedVersionException(t, isFulfillmentStage)) {
Review Comment:
I think we should add a branch like `else if (t instanceof UnsupportedVersionException)`, and return the keys which are not retriable to put into the result using `completeExceptionally(failed)`, and left the others to be retried.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ if (offsetQuery == ListOffsetsRequest.MAX_TIMESTAMP) {
+ if (skipMaxTimestampRequests) {
+ continue;
+ }
+ supportsMaxTimestamp = true;
+ }
+ ListOffsetsTopic topic = topicsByName.computeIfAbsent(
+ topicPartition.topic(), topicName -> new ListOffsetsTopic().setName(topicName));
+ topic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setTimestamp(offsetQuery));
+ }
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ maybeMarkOtherFailures(
Review Comment:
What does this method do? we have already handled all partitions.
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1159516278
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
Review Comment:
I think we should make `supportsMaxTimestamp` a class field and init it in construct method, then in `handleUnsupportedVersion` we will change it to false.
This is what we did in the past in `KafkaAdminClient.getListOffsetsCalls`, right?
https://github.com/apache/kafka/blob/9b409de1e2719f72e3c17663ce131e40dc459361/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L3922-L3944
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
Review Comment:
We translate `OffsetSpec` to long every time we use it, we can just change `offsetSpecsByPartition` to use `Map<TopicPartition, Long>`
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,18 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
+ Map<K, Throwable> unrecoverableFailures =
+ handler.handleUnsupportedVersionException(
+ (UnsupportedVersionException) t,
+ spec.keys,
+ spec.scope instanceof FulfillmentScope);
Review Comment:
Yes, I agree with you, we can remove `isFulfillmentStage` it from `handleUnsupportedVersionException`, and add another `handleUnsupportedVersionException` to `AdminApiLookupStrategy` if necessary in the future, currently, we can just `retryLookup`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/OffsetSpec.java:
##########
@@ -33,7 +33,7 @@ public static class TimestampSpec extends OffsetSpec {
this.timestamp = timestamp;
}
- long timestamp() {
+ public long timestamp() {
Review Comment:
We need to maintain it in the future if we change it public and may bring some compatible problem, we can move `getOffsetFromSpec` to `KafkaAdminClient` to solve it.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170897060
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java:
##########
@@ -70,6 +71,23 @@
*/
ApiResult<K, V> handleResponse(Node broker, Set<K> keys, AbstractResponse response);
+ /**
+ * Callback that is invoked when a fulfillment request hits an UnsupportedVersionException.
+ * Keys for which the exception cannot be handled and the request shouldn't be retried must be mapped
+ * to an error and returned. The request will then be retried for the remainder of the keys.
+ *
+ * @return The failure mappings for the keys for which the exception cannot be handled and the
+ * request shouldn't be retried. If the exception cannot be handled all initial keys will be in
+ * the returned map.
+ */
+ default Map<K, Throwable> handleUnsupportedVersionException(
+ int brokerId,
Review Comment:
That's what I first looked into but I didn't find a nice way to achieve it unless I also add a `Node` param in `onFailure` and pass down the current node from the `Call` instance in `KafkaAdminClient.newCall`. That however would require a number of changes related to the new `onFailure` signature so I am not too fond of the idea.
Let me know if you have anything better in mind.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1153335828
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ if (offsetQuery == ListOffsetsRequest.MAX_TIMESTAMP) {
+ if (skipMaxTimestampRequests) {
+ continue;
+ }
+ supportsMaxTimestamp = true;
+ }
+ ListOffsetsTopic topic = topicsByName.computeIfAbsent(
+ topicPartition.topic(), topicName -> new ListOffsetsTopic().setName(topicName));
+ topic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setTimestamp(offsetQuery));
+ }
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ maybeMarkOtherFailures(
+ broker,
+ keys,
+ failed,
+ p -> !unmapped.isEmpty() || completed.containsKey(p) || failed.containsKey(p) || retriable.contains(p));
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ private void maybeMarkOtherFailures(
+ Node broker,
+ Set<TopicPartition> keys,
+ Map<TopicPartition, Throwable> failed,
+ Predicate<TopicPartition> sanityCheck
+ ) {
+ Exception unsupportedMaxTimestampSpecException = new UnsupportedVersionException(
+ "Broker " + broker.id() + " does not support MAX_TIMESTAMP offset spec");
+ for (TopicPartition topicPartition : keys) {
+ // Fail any MAX_TIMESTAMP requests if we had to downgrade the original request
+ if (skipMaxTimestampRequests
+ && getOffsetFromSpec(
+ offsetSpecsByPartition.get(topicPartition)) == ListOffsetsRequest.MAX_TIMESTAMP
+ ) {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to unsupported version",
+ topicPartition,
+ unsupportedMaxTimestampSpecException);
+ failed.put(topicPartition, unsupportedMaxTimestampSpecException);
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ } else if (!sanityCheck.test(topicPartition)) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ unsupportedMaxTimestampSpecException);
Review Comment:
The wrong exception is being logged here, should be `sanityCheckException`.
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1167773347
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,29 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
Review Comment:
We should add 2 test cases in AdminApiDriverTest to test this:
1. testFulfillmentUnsupportedVersion
2. testFulfillmentRetribleUnsupportedVersion
The first is to verify unsupportedVersionException will return in failed, the code like this:
```
ctx.poll(emptyMap(), map(mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api"))));
ctx.poll(emptyMap(), emptyMap());
```
The second is more complex, you should implement `handleUnsupportedVersion` in `MockAdminApiHandler` to support retry on unsupported, the code may like this:
```
ctx.handler.addRetriableUnsupportedVersionKey("foo");
ctx.handler.expectRequest(mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api")));
ctx.handler.expectRequest(mkSet("bar"), failed("foo", new UnsupportedVersionException("unsupported api")));
// verify retry on foo and not retry on bar.
List<RequestSpec<String>> requestSpecs = ctx.driver.poll();
requestSpecs.forEach(requestSpec -> {
ctx.driver.onFailure(ctx.time.milliseconds(), requestSpec, new UnsupportedVersionException("unsupported api"));
});
ctx.poll(emptyMap(), map(
mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api"))
));
```
--
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] dimitarndimitrov commented on pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on PR #13432:
URL: https://github.com/apache/kafka/pull/13432#issuecomment-1514934648
> please also change the description to include the modifications you made
@dengziming I have updated the PR description a bit and added a "review modifications" section. I assume I shouldn't squash the commits now to preserve the commit history in the PR.
Let me know if there's anything else I need to do regarding that.
I'm now looking at the latest PR CI results.
There seems to be a significant number of test failures, a big increase over the results from previous commits. I suspect they are related to the recent mailing list thread about test failures but am still verifying 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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170758199
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
Review Comment:
I rechecked the old logic of `KafkaAdminClient.getListOffsetsCalls`, we firstly find the leader of each partition and invoke `completeExceptionally` if a partition are not bind to a leader, there is no logic of retrying lookup a leader for a partition, so it's OK even not to retry lookup, right?
I think we should be careful to retry from lookup after inspecting all possible exceptions in `KafkaApis.handleListOffsetRequest`, it's only necessary to retry on`NotLeaderOrFollowerException` and `LeaderNotAvailableException`, and for other `MetadataException`, for example, `KafkaStorageException`, it's nonsense to retry.
In the long run, we can maintain this information in each `lookupStrategy`, but for now, we just retry on these 2 cases similar to `DescribeProducersHandler` and `DescribeTransactionsHandler`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java:
##########
@@ -70,6 +71,23 @@
*/
ApiResult<K, V> handleResponse(Node broker, Set<K> keys, AbstractResponse response);
+ /**
+ * Callback that is invoked when a fulfillment request hits an UnsupportedVersionException.
+ * Keys for which the exception cannot be handled and the request shouldn't be retried must be mapped
+ * to an error and returned. The request will then be retried for the remainder of the keys.
+ *
+ * @return The failure mappings for the keys for which the exception cannot be handled and the
+ * request shouldn't be retried. If the exception cannot be handled all initial keys will be in
+ * the returned map.
+ */
+ default Map<K, Throwable> handleUnsupportedVersionException(
+ int brokerId,
Review Comment:
nit: use Node to keep it in consistent with `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] dajac commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1154222435
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,18 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
+ Map<K, Throwable> unrecoverableFailures =
+ handler.handleUnsupportedVersionException(
+ (UnsupportedVersionException) t,
+ spec.keys,
+ spec.scope instanceof FulfillmentScope);
Review Comment:
If the scope is NOT a FulfillmentScope, I think that we don't have to `completeExceptionally` but rather to `retryLookup` like we do in the previous branch. Am I wrong? If this is true, it may be better to completely separate the two cases. For instance, we could also add `handleUnsupportedVersionException` to the lookup strategy in the future.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163925933
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
Review Comment:
Done.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1171490648
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.LeaderNotAvailableException;
+import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
Review Comment:
@dengziming I've replaced `error.exception()` with the `Errors` instance in all logging here as well as in the comparison for the first case. I have to continue using `error.exception()` for the `RetriableException` matching (and the `failed` map).
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170905273
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
Review Comment:
Done.
> In the long run, we can maintain this information in each lookupStrategy, but for now, we just retry on these 2 cases similar to DescribeProducersHandler and DescribeTransactionsHandler.
Wouldn't we have to keep that code in the handlers as well, as the metadata changes might have happened after the lookup stage has completed? Or do you mean abstracting the handling of failures that require unmapping in a common method of the lookup strategy?
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170012160
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys
Review Comment:
Done, logged at `WARN`, let me know if something different is preferable.
I also considered whether something should be logged in the default method but decided that it's probably better to avoid that as the default handling is to fail all keys which I assume would be logged later somewhere.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1169819833
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4579,7 +4374,7 @@ void maybeRetry(long currentTimeMs, Throwable throwable) {
};
}
- private long getOffsetFromOffsetSpec(OffsetSpec offsetSpec) {
+ private static long getOffsetFromSpec(OffsetSpec offsetSpec) {
Review Comment:
Preface: I am totally fine with reverting these if you prefer so.
Having said that, can I use this as an opportunity to educate myself on preferred style in AK? Let me know if it's preferable to avoid minor code changes like these if the justifications for the change are mostly stylistic, like mine below:
I added the `static` modifier as it indicates that the corresponding method doesn't read or mutate instance state and in this specific case (where class state is also not affected) it's a relatively simple utility method. For a class with as many other methods as `KafkaAdminClient` this seemed useful.
I shortened the name as the current one seemed tautological and because the sole usage of the method is in a bit of a wordy expression to begin with:
```
Map<TopicPartition, Long> offsetQueriesByPartition = topicPartitionOffsets.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e -> getOffsetFromSpec(e.getValue())));
```
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163915187
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,18 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
+ Map<K, Throwable> unrecoverableFailures =
+ handler.handleUnsupportedVersionException(
+ (UnsupportedVersionException) t,
+ spec.keys,
+ spec.scope instanceof FulfillmentScope);
Review Comment:
Yeah, that makes sense.
I initially thought about this but didn't want to expand on it too much because then it might also make sense to compact it with the `NoBatchedFindCoordinatorsException`/`NoBatchedOffsetFetchRequestException` case slightly above, so I was concerned that that many changes unrelated to the ListOffsets API might be undesirable.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163925697
##########
clients/src/main/java/org/apache/kafka/clients/admin/OffsetSpec.java:
##########
@@ -33,7 +33,7 @@ public static class TimestampSpec extends OffsetSpec {
this.timestamp = timestamp;
}
- long timestamp() {
+ public long timestamp() {
Review Comment:
Done.
--
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] dimitarndimitrov commented on pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on PR #13432:
URL: https://github.com/apache/kafka/pull/13432#issuecomment-1481689669
After a rerun, the CI this time failed on `CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated` and `MirrorConnectorsWithCustomForwardingAdminIntegrationTest.testCreatePartitionsUseProvidedForwardingAdmin`.
Like with the previous failing test, I didn't see the `listOffsets(Map<TopicPartition, OffsetSpec>, ListOffsetsOptions)` API being triggered when debugging either of these, but unlike the previous failing test I was unable to locally reproduce a failure over 500+ runs for the former and over 50+ runs for the latter (which is much slower, hence only 50 runs):
```
> Configure project :
Starting build with version 3.5.0-SNAPSHOT (commit id 3f77d983) using Gradle 8.0.2, Java 14 and Scala 2.13.10
Build properties: maxParallelForks=16, maxScalacThreads=8, maxTestRetries=0
> Task :trogdor:test
Gradle Test Run :trogdor:test > Gradle Test Executor 527 > CoordinatorTest > testTaskRequestWithOldStartMsGetsUpdated() PASSED
BUILD SUCCESSFUL in 4s
19 actionable tasks: 1 executed, 18 up-to-date
Completed run: 505
...
> Configure project :
Starting build with version 3.5.0-SNAPSHOT (commit id 3f77d983) using Gradle 8.0.2, Java 14 and Scala 2.13.10
Build properties: maxParallelForks=16, maxScalacThreads=8, maxTestRetries=0
> Task :connect:mirror:test
Gradle Test Run :connect:mirror:test > Gradle Test Executor 533 > MirrorConnectorsWithCustomForwardingAdminIntegrationTest > testCreatePartitionsUseProvidedForwardingAdmin() PASSED
BUILD SUCCESSFUL in 1m 3s
80 actionable tasks: 1 executed, 79 up-to-date
Completed run: 53
```
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163924668
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> getOffsetFromSpec(offsetSpecsByPartition.get(key)) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys, boolean isFulfillmentStage
+ ) {
+ // An UnsupportedVersionException can be addressed only in the fulfillment stage...
+ if (!isFulfillmentStage) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
+ }
+ Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ boolean containsNonMaxTimestampSpec = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ if (getOffsetFromSpec(offsetSpec) == ListOffsetsRequest.MAX_TIMESTAMP) {
+ maxTimestampPartitions.put(topicPartition, exception);
+ } else {
+ containsNonMaxTimestampSpec = true;
+ }
+ }
+ // ...only if it's caused by a MAX_TIMESTAMP spec for some partition, and only if there is at least
+ // one other partition with a non-MAX_TIMESTAMP spec for which the fulfillment stage can be retried.
+ if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
Review Comment:
I'm not sure I get the "retrying infinitely" part. Can you please elaborate a bit more?
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1171232959
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.LeaderNotAvailableException;
+import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
Review Comment:
Here I recommend to use `Errors` directly and not to use `error.exception()`, `error.exception()` will use `defaultExceptionString`, here we have more information, you can refer to `ListConsumerGroupOffsetsHandler.handleGroupError` as an exapmle.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java:
##########
@@ -579,7 +579,9 @@ public void endOffsetsShouldFailWithTimeoutExceptionWhenTimeoutErrorOccurs() {
Set<TopicPartition> tps = Collections.singleton(tp1);
Long offset = null; // response should use error
Cluster cluster = createCluster(1, topicName, 1);
- try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(new MockTime(), cluster)) {
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(
Review Comment:
We should make this change because we will retry on receiving `RetriableException` whereas in the past we only retry on receiving `InvalidMetadataException`, this is a reasonable change.
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1167684205
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4579,7 +4374,7 @@ void maybeRetry(long currentTimeMs, Throwable throwable) {
};
}
- private long getOffsetFromOffsetSpec(OffsetSpec offsetSpec) {
+ private static long getOffsetFromSpec(OffsetSpec offsetSpec) {
Review Comment:
nit: this change seems unnecessary.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,29 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
Review Comment:
We should add 2 test cases in AdminApiDriverTest to test this:
1. testFulfillmentUnsupportedVersion
2. testFulfillmentRetribleUnsupportedVersion
The first is to verify unsupportedVersionException will return in failed, the code like this:
```
ctx.poll(emptyMap(), map(mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api"))));
ctx.poll(emptyMap(), emptyMap());
```
The second is more complex, you should implement `handleUnsupportedVersion` in `MockAdminApiHandler` to support retry on unsupported, the code may like this:
```
ctx.handler.addRetriableUnsupportedVersionKey("foo");
ctx.handler.expectRequest(mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api")));
ctx.handler.expectRequest(mkSet("bar"), failed("foo", new UnsupportedVersionException("unsupported api")));
// verify retry on foo and not retry on bar.
List<RequestSpec<String>> requestSpecs = ctx.driver.poll();
requestSpecs.forEach(requestSpec -> {
ctx.driver.onFailure(ctx.time.milliseconds(), requestSpec, new UnsupportedVersionException("unsupported api"));
});
ctx.poll(emptyMap(), map(
mkSet("foo"), failed("foo", new UnsupportedVersionException("unsupported api"))
));
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> getOffsetFromSpec(offsetSpecsByPartition.get(key)) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys, boolean isFulfillmentStage
+ ) {
+ // An UnsupportedVersionException can be addressed only in the fulfillment stage...
+ if (!isFulfillmentStage) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
+ }
+ Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ boolean containsNonMaxTimestampSpec = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ if (getOffsetFromSpec(offsetSpec) == ListOffsetsRequest.MAX_TIMESTAMP) {
+ maxTimestampPartitions.put(topicPartition, exception);
+ } else {
+ containsNonMaxTimestampSpec = true;
+ }
+ }
+ // ...only if it's caused by a MAX_TIMESTAMP spec for some partition, and only if there is at least
+ // one other partition with a non-MAX_TIMESTAMP spec for which the fulfillment stage can be retried.
+ if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
Review Comment:
Aha, I was mislead by the complex logic here since it's not straightforward. we can replace it with `boolean containsNonMaxTimestampSpec = maxTimestampPartitions.size() != keys.size()` or other simpler expression, so `if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec)` can be changed to `if (maxTimestampPartitions.isEmpty() || maxTimestampPartitions.size()==keys.size())`.
And I think we can remove `containsNonMaxTimestampSpec`, if `containsNonMaxTimestampSpec` is false, then the `keys.stream().collect(Collectors.toMap(k -> k, k -> exception))` and `maxTimestampPartitions` are the same.
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, Errors.REQUEST_TIMED_OUT.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Timeouts should be retried within the fulfillment stage as they are a common type of
+ // retriable error.
+ Set<TopicPartition> retriable = singleton(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(retriable);
+ assertResult(result, completed, emptyMap(), emptyList(), retriable);
+ }
+
+ @Test
+ public void testHandlePartitionInvalidMetadataResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.NOT_LEADER_OR_FOLLOWER;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Invalid metadata errors should be retried from the lookup stage as the partition-to-leader
+ // mappings should be recalculated.
+ List<TopicPartition> unmapped = new ArrayList<>();
+ unmapped.add(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(unmapped);
+ assertResult(result, completed, emptyMap(), unmapped, emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionErrorResponse() {
Review Comment:
Ditto, We should change method name to testHandleUnexpectedXXX to be more clear.
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
Review Comment:
We should change method name to testHandleRetrialbeXXX to be more clear.
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
Review Comment:
nit: allowedVersion check is not the coverage of this PR, I think we can consolidate these 3 BuildRequest-tests in one test.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
Review Comment:
It's a little radical to retry on all kinds of InvalidMetadataException, but it's OK to do this to avoid too much if else.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, Long> offsetTimestampsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetTimestampsByPartition = offsetTimestampsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetTimestamp));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ if (!offsetTimestampsByPartition.containsKey(topicPartition)) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys
Review Comment:
I think we can also add broker to this method, and add some logs such as: "Broker " + broker + " does not support MAX_TIMESTAMP offset spec".
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, Errors.REQUEST_TIMED_OUT.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Timeouts should be retried within the fulfillment stage as they are a common type of
+ // retriable error.
+ Set<TopicPartition> retriable = singleton(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(retriable);
+ assertResult(result, completed, emptyMap(), emptyList(), retriable);
+ }
+
+ @Test
+ public void testHandlePartitionInvalidMetadataResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.NOT_LEADER_OR_FOLLOWER;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Invalid metadata errors should be retried from the lookup stage as the partition-to-leader
+ // mappings should be recalculated.
+ List<TopicPartition> unmapped = new ArrayList<>();
+ unmapped.add(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(unmapped);
+ assertResult(result, completed, emptyMap(), unmapped, emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionErrorResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.UNKNOWN_SERVER_ERROR;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ failed.put(errorPartition, error.exception());
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(failed.keySet());
+ assertResult(result, completed, failed, emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandleResponseSanityCheck() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Long> specsByPartition = new HashMap<>(offsetTimestampsByPartition);
+ specsByPartition.remove(errorPartition);
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap(), specsByPartition));
+
+ assertEquals(offsetTimestampsByPartition.size() - 1, result.completedKeys.size());
+ assertEquals(1, result.failedKeys.size());
+ assertEquals(errorPartition, result.failedKeys.keySet().iterator().next());
+ String sanityCheckMessage = result.failedKeys.get(errorPartition).getMessage();
+ assertTrue(sanityCheckMessage.contains("did not contain a result for topic partition"));
+ assertTrue(result.unmappedKeys.isEmpty());
+ }
+
+ @Test
+ public void testHandleResponseUnsupportedVersion() {
+ UnsupportedVersionException uve = new UnsupportedVersionException("");
+ Map<TopicPartition, OffsetSpec> maxTimestampPartitions = new HashMap<>();
+ maxTimestampPartitions.put(t1p1, OffsetSpec.maxTimestamp());
+
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+
+ final Map<TopicPartition, Long> nonMaxTimestampPartitions = new HashMap<>(offsetTimestampsByPartition);
+ maxTimestampPartitions.forEach((k, v) -> nonMaxTimestampPartitions.remove(k));
+ // Unsupported version exceptions currently cannot be handled if there's no partition with a
+ // MAX_TIMESTAMP spec...
+ Set<TopicPartition> keysToTest = nonMaxTimestampPartitions.keySet();
+ Set<TopicPartition> expectedFailures = keysToTest;
+ assertEquals(
+ mapToError(expectedFailures, uve),
+ handler.handleUnsupportedVersionException(uve, keysToTest));
+
+ // ...or if there are only partitions with MAX_TIMESTAMP specs...
+ keysToTest = maxTimestampPartitions.keySet();
+ expectedFailures = keysToTest;
+ assertEquals(
+ mapToError(expectedFailures, uve),
+ handler.handleUnsupportedVersionException(uve, keysToTest));
+
+ keysToTest = offsetTimestampsByPartition.keySet();
Review Comment:
nit: add a comment about this part, for example, xxx partitions both MAX_TIMESTAMP and not.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
Review Comment:
The original logic is a little messy since we are trying to maintain retriable state using `handleUnsupportedVersionException`, but return value is only boolean, so we add `supportsMaxTimestamp` to maintain more information, and the logic will become more messy if we upgrade it once more.
Your change makes sense to me.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> getOffsetFromSpec(offsetSpecsByPartition.get(key)) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys, boolean isFulfillmentStage
+ ) {
+ // An UnsupportedVersionException can be addressed only in the fulfillment stage...
+ if (!isFulfillmentStage) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
+ }
+ Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ boolean containsNonMaxTimestampSpec = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ if (getOffsetFromSpec(offsetSpec) == ListOffsetsRequest.MAX_TIMESTAMP) {
+ maxTimestampPartitions.put(topicPartition, exception);
+ } else {
+ containsNonMaxTimestampSpec = true;
+ }
+ }
+ // ...only if it's caused by a MAX_TIMESTAMP spec for some partition, and only if there is at least
+ // one other partition with a non-MAX_TIMESTAMP spec for which the fulfillment stage can be retried.
+ if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
Review Comment:
To clarify, there are 3 situations we would meet here:
1. keys are all non-MAX_TIMESTAMP, then all keys will fail directly;
2. keys are all MAX_TIMESTAMP, then all keys will fail directly;
3. keys are mixture of them, MAX_TIMESTAMP keys will fail directly and non-MAX_TIMESTAMP will left to retry(and will fail in the next round if it's still unsupported).
This is the same with the original logic:
1. keys are all non-MAX_TIMESTAMP, supportsMaxTimestamp=false, so all keys will fail directly.
2. keys are all MAX_TIMESTAMP, then all keys will fail directly whether supportsMaxTimestamp Is true or false.
3. keys are mixture, supportsMaxTimestamp=true, MAX_TIMESTAMP keys will fail directly and non-MAX_TIMESTAMP will left to retry(and will fail in the next round if it's still unsupported since supportsMaxTimestamp=false).
The new logic will be clearer.
--
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] dengziming merged pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming merged PR #13432:
URL: https://github.com/apache/kafka/pull/13432
--
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] dengziming commented on pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on PR #13432:
URL: https://github.com/apache/kafka/pull/13432#issuecomment-1515639065
Notable changes better than old logic:
1. Retry lookup stage on receiving `NOT_LEADER_OR_FOLLOWER` and `LEADER_NOT_AVAILABLE`, whereas in the past we failed the partition directly without retry.
2. Removing class field `supportsMaxTimestamp` and calculating it on the fly to avoid the mutable state, this won't change any behavior of the client.
3. Retry fulfillment stage on `RetriableException`, whereas in the past we just retry fulfillment stage on `InvalidMetadataException`, this means we will retry on `TimeoutException` and other `RetriableException`.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1169824036
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
Review Comment:
Done.
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, Errors.REQUEST_TIMED_OUT.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Timeouts should be retried within the fulfillment stage as they are a common type of
+ // retriable error.
+ Set<TopicPartition> retriable = singleton(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(retriable);
+ assertResult(result, completed, emptyMap(), emptyList(), retriable);
+ }
+
+ @Test
+ public void testHandlePartitionInvalidMetadataResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.NOT_LEADER_OR_FOLLOWER;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Invalid metadata errors should be retried from the lookup stage as the partition-to-leader
+ // mappings should be recalculated.
+ List<TopicPartition> unmapped = new ArrayList<>();
+ unmapped.add(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(unmapped);
+ assertResult(result, completed, emptyMap(), unmapped, emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionErrorResponse() {
Review Comment:
Done.
--
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] dimitarndimitrov commented on pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on PR #13432:
URL: https://github.com/apache/kafka/pull/13432#issuecomment-1480931977
The CI run has failed due to a timeout in `kafka.api.PlaintextAdminIntegrationTest.testReplicaCanFetchFromLogStartOffsetAfterDeleteRecords`. Locally I was able to occasionally reproduce the failure on both my PR branch and on `trunk` ~1 out of 10 times. I also read the test code and debugged it and couldn't see the `listOffsets(Map<TopicPartition, OffsetSpec>, ListOffsetsOptions)` API being triggered so I don't think the failure is related to my changes.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1169835470
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
+ ListOffsetsHandler defaultOptionsHandler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest.Builder builder =
+ defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(1, builder.oldestAllowedVersion());
+
+ ListOffsetsHandler readCommittedHandler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
+ assertEquals(2, builder.oldestAllowedVersion());
+
+ builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
+ assertEquals(7, builder.oldestAllowedVersion());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap()));
+
+ assertResult(result, offsetTimestampsByPartition.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionTimeoutResponse() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, Errors.REQUEST_TIMED_OUT.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Timeouts should be retried within the fulfillment stage as they are a common type of
+ // retriable error.
+ Set<TopicPartition> retriable = singleton(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(retriable);
+ assertResult(result, completed, emptyMap(), emptyList(), retriable);
+ }
+
+ @Test
+ public void testHandlePartitionInvalidMetadataResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.NOT_LEADER_OR_FOLLOWER;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Invalid metadata errors should be retried from the lookup stage as the partition-to-leader
+ // mappings should be recalculated.
+ List<TopicPartition> unmapped = new ArrayList<>();
+ unmapped.add(errorPartition);
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(unmapped);
+ assertResult(result, completed, emptyMap(), unmapped, emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionErrorResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.UNKNOWN_SERVER_ERROR;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ failed.put(errorPartition, error.exception());
+ Set<TopicPartition> completed = new HashSet<>(offsetTimestampsByPartition.keySet());
+ completed.removeAll(failed.keySet());
+ assertResult(result, completed, failed, emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandleResponseSanityCheck() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Long> specsByPartition = new HashMap<>(offsetTimestampsByPartition);
+ specsByPartition.remove(errorPartition);
+
+ ApiResult<TopicPartition, ListOffsetsResultInfo> result =
+ handleResponse(createResponse(emptyMap(), specsByPartition));
+
+ assertEquals(offsetTimestampsByPartition.size() - 1, result.completedKeys.size());
+ assertEquals(1, result.failedKeys.size());
+ assertEquals(errorPartition, result.failedKeys.keySet().iterator().next());
+ String sanityCheckMessage = result.failedKeys.get(errorPartition).getMessage();
+ assertTrue(sanityCheckMessage.contains("did not contain a result for topic partition"));
+ assertTrue(result.unmappedKeys.isEmpty());
+ }
+
+ @Test
+ public void testHandleResponseUnsupportedVersion() {
+ UnsupportedVersionException uve = new UnsupportedVersionException("");
+ Map<TopicPartition, OffsetSpec> maxTimestampPartitions = new HashMap<>();
+ maxTimestampPartitions.put(t1p1, OffsetSpec.maxTimestamp());
+
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+
+ final Map<TopicPartition, Long> nonMaxTimestampPartitions = new HashMap<>(offsetTimestampsByPartition);
+ maxTimestampPartitions.forEach((k, v) -> nonMaxTimestampPartitions.remove(k));
+ // Unsupported version exceptions currently cannot be handled if there's no partition with a
+ // MAX_TIMESTAMP spec...
+ Set<TopicPartition> keysToTest = nonMaxTimestampPartitions.keySet();
+ Set<TopicPartition> expectedFailures = keysToTest;
+ assertEquals(
+ mapToError(expectedFailures, uve),
+ handler.handleUnsupportedVersionException(uve, keysToTest));
+
+ // ...or if there are only partitions with MAX_TIMESTAMP specs...
+ keysToTest = maxTimestampPartitions.keySet();
+ expectedFailures = keysToTest;
+ assertEquals(
+ mapToError(expectedFailures, uve),
+ handler.handleUnsupportedVersionException(uve, keysToTest));
+
+ keysToTest = offsetTimestampsByPartition.keySet();
Review Comment:
Done.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1169851050
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandlerTest.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.admin.internals;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+public final class ListOffsetsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+
+ private final Node node = new Node(1, "host", 1234);
+
+ private final Map<TopicPartition, Long> offsetTimestampsByPartition = new HashMap<TopicPartition, Long>() {
+ {
+ put(t0p0, ListOffsetsRequest.LATEST_TIMESTAMP);
+ put(t0p1, ListOffsetsRequest.EARLIEST_TIMESTAMP);
+ put(t1p0, 123L);
+ put(t1p1, ListOffsetsRequest.MAX_TIMESTAMP);
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
+ ListOffsetsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(1, topics.size());
+ ListOffsetsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ assertExpectedTimestamp(topicPartition, partition.timestamp());
+ }
+ assertEquals(IsolationLevel.READ_UNCOMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestMultipleTopicsWithReadCommitted() {
+ ListOffsetsHandler handler =
+ new ListOffsetsHandler(
+ offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
+ ListOffsetsRequest request =
+ handler.buildBatchedRequest(node.id(), offsetTimestampsByPartition.keySet()).build();
+ List<ListOffsetsTopic> topics = request.topics();
+ assertEquals(2, topics.size());
+ Map<TopicPartition, ListOffsetsPartition> partitions = new HashMap<>();
+ for (ListOffsetsTopic topic : topics) {
+ for (ListOffsetsPartition partition : topic.partitions()) {
+ partitions.put(new TopicPartition(topic.name(), partition.partitionIndex()), partition);
+ }
+ }
+ assertEquals(4, partitions.size());
+ for (Map.Entry<TopicPartition, ListOffsetsPartition> entry : partitions.entrySet()) {
+ assertExpectedTimestamp(entry.getKey(), entry.getValue().timestamp());
+ }
+ assertEquals(IsolationLevel.READ_COMMITTED, request.isolationLevel());
+ }
+
+ @Test
+ public void testBuildRequestAllowedVersions() {
Review Comment:
> allowedVersion check is not the coverage of this PR
I think in a way it is, as the newly introduced `AdminApiDriver` mechanism does affect how the allowed version is computed.
> I think we can consolidate these 3 BuildRequest-tests in one test.
If you don't insist strongly, I would prefer to keep these as-is - cramming them altogether will make for a bit of a hard to read super-test. There are other handler tests with multiple request building tests, like the latest addition - `ListConsumerGroupOffsetsHandlerTest` - so this won't be too much of a precedent.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170437027
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -260,12 +261,29 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+ } else if (t instanceof UnsupportedVersionException) {
Review Comment:
Done.
Also added a check that `UnsupportedVersionException` during the lookup stage causes a failure.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1170897060
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java:
##########
@@ -70,6 +71,23 @@
*/
ApiResult<K, V> handleResponse(Node broker, Set<K> keys, AbstractResponse response);
+ /**
+ * Callback that is invoked when a fulfillment request hits an UnsupportedVersionException.
+ * Keys for which the exception cannot be handled and the request shouldn't be retried must be mapped
+ * to an error and returned. The request will then be retried for the remainder of the keys.
+ *
+ * @return The failure mappings for the keys for which the exception cannot be handled and the
+ * request shouldn't be retried. If the exception cannot be handled all initial keys will be in
+ * the returned map.
+ */
+ default Map<K, Throwable> handleUnsupportedVersionException(
+ int brokerId,
Review Comment:
That's what I first looked into but I didn't find a nice way to achieve it unless I also add a `Node` param in `onFailure` and pass down the current node from the `Call` instance in `KafkaAdminClient.newCall`. That however would require a number of changes related to the new `onFailure` signature so I am not to fond of the idea.
Let me know if you have anything better in mind.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1153153515
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ if (offsetQuery == ListOffsetsRequest.MAX_TIMESTAMP) {
+ if (skipMaxTimestampRequests) {
Review Comment:
I opted out of that or something similar to avoid a separate O(N) iteration. I think it can still be avoided even with `CollectionUtils.groupPartitionsByTopic` with a boolean-pointer-equivalent (e.g. `AtomicBoolean`). Still I'll keep it simple for now and can add that if you think it's preferable.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ if (offsetQuery == ListOffsetsRequest.MAX_TIMESTAMP) {
+ if (skipMaxTimestampRequests) {
+ continue;
+ }
+ supportsMaxTimestamp = true;
+ }
+ ListOffsetsTopic topic = topicsByName.computeIfAbsent(
+ topicPartition.topic(), topicName -> new ListOffsetsTopic().setName(topicName));
+ topic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setTimestamp(offsetQuery));
+ }
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ maybeMarkOtherFailures(
Review Comment:
It handles `MAX_TIMESTAMP` partitions (which are not eagerly failed in `handleUnsupportedVersionException` but just omitted from the next request) and sanity-checks all partitions (the equivalent of [this check](https://github.com/apache/kafka/blob/9b409de1e2719f72e3c17663ce131e40dc459361/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L3892-L3900)).
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -261,19 +262,30 @@ public void onFailure(
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
} else {
+ boolean isFulfillmentStage = spec.scope instanceof FulfillmentScope;
+ if (handleUnsupportedVersionException(t, isFulfillmentStage)) {
Review Comment:
That's neat, let me fix it.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1154040840
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java:
##########
@@ -70,6 +71,22 @@
*/
ApiResult<K, V> handleResponse(Node broker, Set<K> keys, AbstractResponse response);
+ /**
+ * Callback that is invoked when a request hits an UnsupportedVersionException. If the
+ * exception can be handled and the request retried, the appropriate changes to the request
+ * should be applied and that should be indicated via the returned boolean value. If the
+ * exception is thrown during the lookup stage the handler should delegate the handling to the
+ * lookup strategy.
+ *
+ * @return True if the exception can be handled; false otherwise.
+ */
+ default boolean handleUnsupportedVersionException(
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final ListOffsetsOptions options;
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ private volatile boolean skipMaxTimestampRequests = false;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ final Map<String, ListOffsetsTopic> topicsByName = new HashMap<>();
+ boolean supportsMaxTimestamp = false;
+ for (TopicPartition topicPartition : keys) {
Review Comment:
Done.
--
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] dengziming commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dengziming (via GitHub)" <gi...@apache.org>.
dengziming commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1159538238
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
+ .stream()
+ .anyMatch(key -> getOffsetFromSpec(offsetSpecsByPartition.get(key)) == ListOffsetsRequest.MAX_TIMESTAMP);
+
+ return ListOffsetsRequest.Builder
+ .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp)
+ .setTargetTimes(new ArrayList<>(topicsByName.values()));
+ }
+
+ @Override
+ public ApiResult<TopicPartition, ListOffsetsResultInfo> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ ListOffsetsResponse response = (ListOffsetsResponse) abstractResponse;
+ Map<TopicPartition, ListOffsetsResultInfo> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (ListOffsetsTopicResponse topic : response.topics()) {
+ for (ListOffsetsPartitionResponse partition : topic.partitions()) {
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
+ Errors error = Errors.forCode(partition.errorCode());
+ OffsetSpec offsetRequestSpec = offsetSpecsByPartition.get(topicPartition);
+ if (offsetRequestSpec == null) {
+ log.warn("ListOffsets response includes unknown topic partition {}", topicPartition);
+ } else if (error == Errors.NONE) {
+ Optional<Integer> leaderEpoch = (partition.leaderEpoch() == ListOffsetsResponse.UNKNOWN_EPOCH)
+ ? Optional.empty()
+ : Optional.of(partition.leaderEpoch());
+ completed.put(
+ topicPartition,
+ new ListOffsetsResultInfo(partition.offset(), partition.timestamp(), leaderEpoch));
+ } else {
+ handlePartitionError(topicPartition, error, failed, unmapped, retriable);
+ }
+ }
+ }
+
+ // Sanity-check if the current leader for these partitions returned results for all of them
+ for (TopicPartition topicPartition : keys) {
+ if (unmapped.isEmpty()
+ && !completed.containsKey(topicPartition)
+ && !failed.containsKey(topicPartition)
+ && !retriable.contains(topicPartition)
+ ) {
+ ApiException sanityCheckException = new ApiException(
+ "The response from broker " + broker.id() +
+ " did not contain a result for topic partition " + topicPartition);
+ log.error(
+ "ListOffsets request for topic partition {} failed sanity check",
+ topicPartition,
+ sanityCheckException);
+ failed.put(topicPartition, sanityCheckException);
+ }
+ }
+
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handlePartitionError(
+ TopicPartition topicPartition,
+ Errors error,
+ Map<TopicPartition, Throwable> failed,
+ List<TopicPartition> unmapped,
+ Set<TopicPartition> retriable
+ ) {
+ ApiException apiException = error.exception();
+ if (apiException instanceof InvalidMetadataException) {
+ log.debug(
+ "ListOffsets lookup request for topic partition {} will be retried due to invalid metadata",
+ topicPartition,
+ apiException);
+ unmapped.add(topicPartition);
+ } else if (apiException instanceof RetriableException) {
+ log.debug(
+ "ListOffsets fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ apiException);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "ListOffsets request for topic partition {} failed due to an unexpected error",
+ topicPartition,
+ apiException);
+ failed.put(topicPartition, apiException);
+ }
+ }
+
+ @Override
+ public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
+ UnsupportedVersionException exception, Set<TopicPartition> keys, boolean isFulfillmentStage
+ ) {
+ // An UnsupportedVersionException can be addressed only in the fulfillment stage...
+ if (!isFulfillmentStage) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
+ }
+ Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ boolean containsNonMaxTimestampSpec = false;
+ for (TopicPartition topicPartition : keys) {
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ if (getOffsetFromSpec(offsetSpec) == ListOffsetsRequest.MAX_TIMESTAMP) {
+ maxTimestampPartitions.put(topicPartition, exception);
+ } else {
+ containsNonMaxTimestampSpec = true;
+ }
+ }
+ // ...only if it's caused by a MAX_TIMESTAMP spec for some partition, and only if there is at least
+ // one other partition with a non-MAX_TIMESTAMP spec for which the fulfillment stage can be retried.
+ if (maxTimestampPartitions.isEmpty() || !containsNonMaxTimestampSpec) {
+ return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
Review Comment:
If we make `supportsMaxTimestamp` a class field, we should also change it to false here, and if it already equals false, we can avoid retrying it infinitely.
--
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] dimitarndimitrov commented on a diff in pull request #13432: KAFKA-14821 Implement the listOffsets API with AdminApiDriver
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13432:
URL: https://github.com/apache/kafka/pull/13432#discussion_r1163919103
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.admin.internals;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.ListOffsetsOptions;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.OffsetSpec.TimestampSpec;
+import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.Batched;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.InvalidMetadataException;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
+import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsTopic;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
+import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class ListOffsetsHandler extends Batched<TopicPartition, ListOffsetsResultInfo> {
+
+ private final Map<TopicPartition, OffsetSpec> offsetSpecsByPartition;
+ private final ListOffsetsOptions options;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public ListOffsetsHandler(
+ Map<TopicPartition, OffsetSpec> offsetSpecsByPartition,
+ ListOffsetsOptions options,
+ LogContext logContext
+ ) {
+ this.offsetSpecsByPartition = offsetSpecsByPartition;
+ this.options = options;
+ this.log = logContext.logger(ListOffsetsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "listOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ @Override
+ ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
+ keys,
+ topicName -> new ListOffsetsTopic().setName(topicName),
+ (listOffsetsTopic, partitionId) -> {
+ TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
+ OffsetSpec offsetSpec = offsetSpecsByPartition.get(topicPartition);
+ long offsetQuery = getOffsetFromSpec(offsetSpec);
+ listOffsetsTopic.partitions().add(
+ new ListOffsetsPartition()
+ .setPartitionIndex(partitionId)
+ .setTimestamp(offsetQuery));
+ });
+ boolean supportsMaxTimestamp = keys
Review Comment:
I actually am not sure if this is ideal. Once we trim the `MAX_TIMESTAMP` requests on a retry we no longer should be indicating that there are such requests in the `ListOffsetsRequest.Builder`.
--
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