You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "tinaselenge (via GitHub)" <gi...@apache.org> on 2023/05/24 23:56:21 UTC
[GitHub] [kafka] tinaselenge opened a new pull request, #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
tinaselenge opened a new pull request, #13760:
URL: https://github.com/apache/kafka/pull/13760
The retry of fetching metadata is handled by the AdminApiDriver class.
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### 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] tinaselenge commented on pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on PR #13760:
URL: https://github.com/apache/kafka/pull/13760#issuecomment-1582411061
@divijvaidya @showuon @mimaison Thank you very much for reviewing the PR!
I believe I have addressed the comments now. Please let me know if I have missed anything. Thanks.
--
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] showuon merged pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #13760:
URL: https://github.com/apache/kafka/pull/13760
--
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] showuon commented on pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #13760:
URL: https://github.com/apache/kafka/pull/13760#issuecomment-1602323587
@tinaselenge , it looks like some tests are failed: https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-13760/6/
Please also take a look. Thanks.
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1238257723
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2359,31 +2334,22 @@ public void testDeleteRecords() throws Exception {
assertTrue(e0.getCause() instanceof OffsetOutOfRangeException);
}
- // "leader not available" failure on metadata request for partition 2
+ // not authorized to delete records for partition 2
KafkaFuture<DeletedRecords> myTopicPartition2Result = values.get(myTopicPartition2);
try {
myTopicPartition2Result.get();
fail("get() should throw ExecutionException");
} catch (ExecutionException e1) {
- assertTrue(e1.getCause() instanceof LeaderNotAvailableException);
+ assertTrue(e1.getCause() instanceof TopicAuthorizationException);
}
Review Comment:
Ah, I see. Thanks for the explanation.
--
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] divijvaidya commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1214582937
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+
+ System.out.println("Partitions: " + deleteRecords.partitions());
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
+ return new DeleteRecordsRequest.Builder(data);
+ }
+
+
+ @Override
+ public ApiResult<TopicPartition, DeletedRecords> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ DeleteRecordsResponse response = (DeleteRecordsResponse) abstractResponse;
+ Map<TopicPartition, DeletedRecords> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (DeleteRecordsResponseData.DeleteRecordsTopicResult topicResult: response.data().topics()) {
+ for (DeleteRecordsResponseData.DeleteRecordsPartitionResult partitionResult : topicResult.partitions()) {
+ Errors error = Errors.forCode(partitionResult.errorCode());
+ TopicPartition topicPartition = new TopicPartition(topicResult.name(), partitionResult.partitionIndex());
+ if (error == Errors.NONE) {
+ completed.put(topicPartition, new DeletedRecords(partitionResult.lowWatermark()));
+ } 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(
+ "DeleteRecords 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
+ ) {
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.LEADER_NOT_AVAILABLE) {
Review Comment:
instead, could we use `error.exception() instanceof InvalidMetadataException`? That covers all cases where metadata might be a problem. e.g. the current code is missing `UNKNOWN_TOPIC_OR_PARTITION`
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+
+ System.out.println("Partitions: " + deleteRecords.partitions());
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
+ return new DeleteRecordsRequest.Builder(data);
+ }
+
+
+ @Override
+ public ApiResult<TopicPartition, DeletedRecords> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ DeleteRecordsResponse response = (DeleteRecordsResponse) abstractResponse;
+ Map<TopicPartition, DeletedRecords> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (DeleteRecordsResponseData.DeleteRecordsTopicResult topicResult: response.data().topics()) {
+ for (DeleteRecordsResponseData.DeleteRecordsPartitionResult partitionResult : topicResult.partitions()) {
+ Errors error = Errors.forCode(partitionResult.errorCode());
+ TopicPartition topicPartition = new TopicPartition(topicResult.name(), partitionResult.partitionIndex());
+ if (error == Errors.NONE) {
+ completed.put(topicPartition, new DeletedRecords(partitionResult.lowWatermark()));
+ } 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(
+ "DeleteRecords 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
+ ) {
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.LEADER_NOT_AVAILABLE) {
+ log.debug(
+ "DeleteRecords lookup request for topic partition {} will be retried due to invalid leader metadata {}",
+ topicPartition,
+ error);
+ unmapped.add(topicPartition);
+ } else if (error.exception() instanceof RetriableException) {
+ log.debug(
+ "DeleteRecords fulfillment request for topic partition {} will be retried due to {}",
+ topicPartition,
+ error);
+ retriable.add(topicPartition);
+ } else {
+ log.error(
+ "DeleteRecords request for topic partition {} failed due to an unexpected error {}",
Review Comment:
we should probably catch `TOPIC_AUTHORIZATION_FAILED` separately since that is not "unexpected"
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+
+ System.out.println("Partitions: " + deleteRecords.partitions());
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
+ return new DeleteRecordsRequest.Builder(data);
+ }
+
+
+ @Override
+ public ApiResult<TopicPartition, DeletedRecords> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ DeleteRecordsResponse response = (DeleteRecordsResponse) abstractResponse;
+ Map<TopicPartition, DeletedRecords> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (DeleteRecordsResponseData.DeleteRecordsTopicResult topicResult: response.data().topics()) {
+ for (DeleteRecordsResponseData.DeleteRecordsPartitionResult partitionResult : topicResult.partitions()) {
+ Errors error = Errors.forCode(partitionResult.errorCode());
+ TopicPartition topicPartition = new TopicPartition(topicResult.name(), partitionResult.partitionIndex());
+ if (error == Errors.NONE) {
+ completed.put(topicPartition, new DeletedRecords(partitionResult.lowWatermark()));
+ } 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(
+ "DeleteRecords 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
+ ) {
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.LEADER_NOT_AVAILABLE) {
Review Comment:
Please extract out handling of NOT_LEADER_OR_FOLLOWER at a common place for clients: e.g. other usage https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java#L158
--
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] showuon commented on pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #13760:
URL: https://github.com/apache/kafka/pull/13760#issuecomment-1614272121
Re-running CI build: https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-13760/9/
--
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 #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1229356799
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+
+ System.out.println("Partitions: " + deleteRecords.partitions());
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
+ return new DeleteRecordsRequest.Builder(data);
+ }
+
+
+ @Override
+ public ApiResult<TopicPartition, DeletedRecords> handleResponse(
+ Node broker,
+ Set<TopicPartition> keys,
+ AbstractResponse abstractResponse
+ ) {
+ DeleteRecordsResponse response = (DeleteRecordsResponse) abstractResponse;
+ Map<TopicPartition, DeletedRecords> completed = new HashMap<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ List<TopicPartition> unmapped = new ArrayList<>();
+ Set<TopicPartition> retriable = new HashSet<>();
+
+ for (DeleteRecordsResponseData.DeleteRecordsTopicResult topicResult: response.data().topics()) {
+ for (DeleteRecordsResponseData.DeleteRecordsPartitionResult partitionResult : topicResult.partitions()) {
+ Errors error = Errors.forCode(partitionResult.errorCode());
+ TopicPartition topicPartition = new TopicPartition(topicResult.name(), partitionResult.partitionIndex());
+ if (error == Errors.NONE) {
+ completed.put(topicPartition, new DeletedRecords(partitionResult.lowWatermark()));
+ } 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(
+ "DeleteRecords 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
+ ) {
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.LEADER_NOT_AVAILABLE) {
Review Comment:
@divijvaidya FWIW when the somewhat similar process was being applied to the `ListOffsets` API and I had the exact same thought, @dengziming noted that not all types of `InvalidMetadataException` should result in a retry (e.g. `KafkaStorageException`). See https://github.com/apache/kafka/pull/13432#discussion_r1167756758
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1238266317
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.TopicAuthorizationException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.computeIfAbsent(
+ topicPartition.topic(),
+ key -> new DeleteRecordsRequestData.DeleteRecordsTopic().setName(topicPartition.topic())
+ );
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
Review Comment:
No, I don't think they are the same thing. The invokeDriver timeout is set in the client side for client timeout the request. But the `timeoutMs` field in `DeleteRecordsRequest` is to send to the server side to notify the server about the timeout.
REF: https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/DeleteRecordsRequest.json#L39
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1228079251
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -2925,123 +2919,11 @@ void handleFailure(Throwable throwable) {
@Override
public DeleteRecordsResult deleteRecords(final Map<TopicPartition, RecordsToDelete> recordsToDelete,
final DeleteRecordsOptions options) {
+ AdminApiFuture.SimpleAdminApiFuture<TopicPartition, DeletedRecords> future = DeleteRecordsHandler.newFuture(recordsToDelete.keySet());
Review Comment:
nit: `AdminApiFuture.SimpleAdminApiFuture` -> `SimpleAdminApiFuture` should work.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.TopicAuthorizationException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.computeIfAbsent(
+ topicPartition.topic(),
+ key -> new DeleteRecordsRequestData.DeleteRecordsTopic().setName(topicPartition.topic())
+ );
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
Review Comment:
Should we set timeout here like before?
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1222617381
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2291,6 +2289,8 @@ public void testDeleteRecords() throws Exception {
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
Review Comment:
Before looking into the implementation, I don't think this test is testing what we expected. The original comment is commenting around this lines:
https://github.com/apache/kafka/pull/7296/files#diff-5422d10d9a7f4776c6538ae3aea27f24e94cf4ecf5e752040125aca6edc795d3R3671-R3682
And it said, when metadata response (mr) contains error, we just fail the future without retry.
The point is, we want to retry `metadataResponse`, not `deleteRecordResponse`. What I expected is tests like this:
https://github.com/apache/kafka/blob/513e1c641d63c5e15144f9fcdafa1b56c5e5ba09/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java#L5431C11-L5433
Please update the test and make sure it passed.
Thanks.
--
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] tinaselenge commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1237673297
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.TopicAuthorizationException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.computeIfAbsent(
+ topicPartition.topic(),
+ key -> new DeleteRecordsRequestData.DeleteRecordsTopic().setName(topicPartition.topic())
+ );
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+ }
+
+ DeleteRecordsRequestData data = new DeleteRecordsRequestData()
+ .setTopics(new ArrayList<>(deletionsForTopic.values()));
Review Comment:
I could be wrong but I think the timeout specified by the user gets passed to AdminApiDriver (via invokeDriver()) and it handles the timeout for the requests.
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1222617381
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2291,6 +2289,8 @@ public void testDeleteRecords() throws Exception {
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
Review Comment:
Before looking into the implementation, I don't think this test is testing what we expected. The original comment is commenting around this lines:
https://github.com/apache/kafka/pull/7296/files#diff-5422d10d9a7f4776c6538ae3aea27f24e94cf4ecf5e752040125aca6edc795d3R3671-R3682
And it said, when metadata response (mr) contains error, we just fail the future without retry.
And here, the existing metadataResponse also fail without retry: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L2969-L2972
The point is, we want to retry `metadataResponse`, not `deleteRecordResponse`. What I expected is tests like this:
https://github.com/apache/kafka/blob/513e1c641d63c5e15144f9fcdafa1b56c5e5ba09/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java#L5431C11-L5433
Please update the test and make sure it passed.
Thanks.
--
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] showuon commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1229107591
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2359,31 +2334,22 @@ public void testDeleteRecords() throws Exception {
assertTrue(e0.getCause() instanceof OffsetOutOfRangeException);
}
- // "leader not available" failure on metadata request for partition 2
+ // not authorized to delete records for partition 2
KafkaFuture<DeletedRecords> myTopicPartition2Result = values.get(myTopicPartition2);
try {
myTopicPartition2Result.get();
fail("get() should throw ExecutionException");
} catch (ExecutionException e1) {
- assertTrue(e1.getCause() instanceof LeaderNotAvailableException);
+ assertTrue(e1.getCause() instanceof TopicAuthorizationException);
}
Review Comment:
This can be replaced with `assertThrows(TopicAuthorizationException.class, () ->myTopicPartition2Result.get(), "get() should throw ExecutionException" )`
Also, Should we change the comment into `TopicAuthorizationException`? It doesn't throw `ExecutionException` at all.
Same comment applies to other places.
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandlerTest.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.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.junit.jupiter.api.Test;
+
+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;
+
+public class DeleteRecordsHandlerTest {
+ private final LogContext logContext = new LogContext();
+ private final TopicPartition t0p0 = new TopicPartition("t0", 0);
+ private final TopicPartition t0p1 = new TopicPartition("t0", 1);
+ private final Node node = new Node(1, "host", 1234);
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<TopicPartition, RecordsToDelete>() {
+ {
+ put(t0p0, RecordsToDelete.beforeOffset(10L));
+ put(t0p1, RecordsToDelete.beforeOffset(10L));
+ }
+ };
+
+ @Test
+ public void testBuildRequestSimple() {
+ DeleteRecordsHandler handler = new DeleteRecordsHandler(recordsToDelete, logContext);
+ DeleteRecordsRequest request = handler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1)).build();
+ List<DeleteRecordsRequestData.DeleteRecordsTopic> topicPartitions = request.data().topics();
+ assertEquals(1, topicPartitions.size());
+ DeleteRecordsRequestData.DeleteRecordsTopic topic = topicPartitions.get(0);
+ assertEquals(2, topic.partitions().size());
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> result =
+ handleResponse(createResponse(emptyMap(), recordsToDelete.keySet()));
+ assertResult(result, recordsToDelete.keySet(), emptyMap(), emptyList(), emptySet());
+ }
+
+ @Test
+ public void testHandleRetriablePartitionTimeoutResponse() {
+ TopicPartition errorPartition = t0p0;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, Errors.REQUEST_TIMED_OUT.code());
+
+ AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> 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<>(recordsToDelete.keySet());
+ completed.removeAll(retriable);
+ assertResult(result, completed, emptyMap(), emptyList(), retriable);
+ }
+
+ @Test
+ public void testHandleLookupRetriablePartitionInvalidMetadataResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.NOT_LEADER_OR_FOLLOWER;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ // Some 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<>(recordsToDelete.keySet());
+ completed.removeAll(unmapped);
+ assertResult(result, completed, emptyMap(), unmapped, emptySet());
+ }
+
+ @Test
+ public void testHandlePartitionErrorResponse() {
+ TopicPartition errorPartition = t0p0;
+ Errors error = Errors.TOPIC_AUTHORIZATION_FAILED;
+ Map<TopicPartition, Short> errorsByPartition = new HashMap<>();
+ errorsByPartition.put(errorPartition, error.code());
+
+ AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> result =
+ handleResponse(createResponse(errorsByPartition));
+
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+ failed.put(errorPartition, error.exception());
+ Set<TopicPartition> completed = new HashSet<>(recordsToDelete.keySet());
+ completed.removeAll(failed.keySet());
+ assertResult(result, completed, failed, emptyList(), emptySet());
Review Comment:
Could we add a test to verify a mixed case? I.e. a test case contains completed, failed, unmapped, and retriable.
--
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] tinaselenge commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1237642066
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2359,31 +2334,22 @@ public void testDeleteRecords() throws Exception {
assertTrue(e0.getCause() instanceof OffsetOutOfRangeException);
}
- // "leader not available" failure on metadata request for partition 2
+ // not authorized to delete records for partition 2
KafkaFuture<DeletedRecords> myTopicPartition2Result = values.get(myTopicPartition2);
try {
myTopicPartition2Result.get();
fail("get() should throw ExecutionException");
} catch (ExecutionException e1) {
- assertTrue(e1.getCause() instanceof LeaderNotAvailableException);
+ assertTrue(e1.getCause() instanceof TopicAuthorizationException);
}
Review Comment:
We are actually catching ExecutionException and checking the cause of the exception. This would not work because it's not throwing TopicAuthorizationException, but it is the supplied cause for the ExecutionException.
--
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] tinaselenge commented on pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on PR #13760:
URL: https://github.com/apache/kafka/pull/13760#issuecomment-1610960042
Thank you @showuon for the feedback. I think I have addressed the comments. Please let me know there is anything I missed or doesn't look right.
--
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] mimaison commented on a diff in pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #13760:
URL: https://github.com/apache/kafka/pull/13760#discussion_r1222667563
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
+ deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
+ .setPartitionIndex(topicPartition.partition())
+ .setOffset(entry.getValue().beforeOffset()));
+
+ System.out.println("Partitions: " + deleteRecords.partitions());
Review Comment:
Let's remove this debugging statement
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.kafka.clients.admin.DeletedRecords;
+import org.apache.kafka.clients.admin.RecordsToDelete;
+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.RetriableException;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedRecords> {
+
+ private final Map<TopicPartition, RecordsToDelete> recordsToDelete;
+ private final Logger log;
+ private final AdminApiLookupStrategy<TopicPartition> lookupStrategy;
+
+ public DeleteRecordsHandler(
+ Map<TopicPartition, RecordsToDelete> recordsToDelete,
+ LogContext logContext
+ ) {
+ this.recordsToDelete = recordsToDelete;
+ this.log = logContext.logger(DeleteRecordsHandler.class);
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+
+ @Override
+ public String apiName() {
+ return "deleteRecords";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<TopicPartition> lookupStrategy() {
+ return this.lookupStrategy;
+ }
+
+ public static SimpleAdminApiFuture<TopicPartition, DeletedRecords> newFuture(
+ Collection<TopicPartition> topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
+ }
+
+ @Override
+ public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
+ Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
+ for (Map.Entry<TopicPartition, RecordsToDelete> entry: recordsToDelete.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.get(topicPartition.topic());
+ if (deleteRecords == null) {
+ deleteRecords = new DeleteRecordsRequestData.DeleteRecordsTopic()
+ .setName(topicPartition.topic());
+ deletionsForTopic.put(topicPartition.topic(), deleteRecords);
+ }
Review Comment:
Nothing wrong with your logic but we could use `computeIfAbsent` here
```suggestion
DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.computeIfAbsent(
topicPartition.topic(),
key -> new DeleteRecordsRequestData.DeleteRecordsTopic().setName(topicPartition.topic())
);
```
--
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] showuon commented on pull request #13760: KAFKA-8982: Add retry of fetching metadata to Admin.deleteRecords
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #13760:
URL: https://github.com/apache/kafka/pull/13760#issuecomment-1617075103
Failed tests are unrelated.
--
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