You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2024/03/07 15:32:14 UTC
(kafka) branch 3.7 updated: KAFKA-16319: Divide DeleteTopics requests by leader node (#15479)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new 94d38ef36bc KAFKA-16319: Divide DeleteTopics requests by leader node (#15479)
94d38ef36bc is described below
commit 94d38ef36bc171319ef73898970dc5ed0eea6e80
Author: Andrew Schofield <as...@confluent.io>
AuthorDate: Thu Mar 7 15:24:11 2024 +0000
KAFKA-16319: Divide DeleteTopics requests by leader node (#15479)
Reviewers: Reviewers: Mickael Maison <mi...@gmail.com>, Kirk True <ki...@kirktrue.pro>, Daniel Gospodinow <dg...@confluent.io>
---
.../admin/internals/DeleteRecordsHandler.java | 6 +-
.../admin/internals/DeleteRecordsHandlerTest.java | 73 +++++++++++++++++++---
2 files changed, 69 insertions(+), 10 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java
index 2daad226034..9f40d19f00b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandler.java
@@ -79,15 +79,15 @@ public final class DeleteRecordsHandler extends Batched<TopicPartition, DeletedR
@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();
+ for (TopicPartition topicPartition : keys) {
+ RecordsToDelete toDelete = recordsToDelete.get(topicPartition);
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()));
+ .setOffset(toDelete.beforeOffset()));
}
DeleteRecordsRequestData data = new DeleteRecordsRequestData()
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandlerTest.java
index c39747f1fba..58492696c4d 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteRecordsHandlerTest.java
@@ -22,15 +22,24 @@ 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.DeletedRecords;
import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.clients.admin.internals.AdminApiLookupStrategy.LookupResult;
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.message.MetadataResponseData;
+import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition;
+import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic;
+import org.apache.kafka.common.protocol.ApiKeys;
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.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.LogContext;
import org.junit.jupiter.api.Test;
@@ -41,6 +50,7 @@ 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.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class DeleteRecordsHandlerTest {
@@ -50,7 +60,8 @@ public class DeleteRecordsHandlerTest {
private final TopicPartition t0p1 = new TopicPartition("t0", 1);
private final TopicPartition t0p2 = new TopicPartition("t0", 2);
private final TopicPartition t0p3 = new TopicPartition("t0", 3);
- private final Node node = new Node(1, "host", 1234);
+ private final Node node1 = new Node(1, "host", 1234);
+ private final Node node2 = new Node(2, "host", 1235);
private final Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<TopicPartition, RecordsToDelete>() {
{
put(t0p0, RecordsToDelete.beforeOffset(10L));
@@ -63,11 +74,11 @@ public class DeleteRecordsHandlerTest {
@Test
public void testBuildRequestSimple() {
DeleteRecordsHandler handler = new DeleteRecordsHandler(recordsToDelete, logContext, timeout);
- 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(4, topic.partitions().size());
+ DeleteRecordsRequest request = handler.buildBatchedRequest(node1.id(), mkSet(t0p0, t0p1)).build();
+ List<DeleteRecordsRequestData.DeleteRecordsTopic> topics = request.data().topics();
+ assertEquals(1, topics.size());
+ DeleteRecordsRequestData.DeleteRecordsTopic topic = topics.get(0);
+ assertEquals(2, topic.partitions().size());
}
@Test
@@ -199,6 +210,54 @@ public class DeleteRecordsHandlerTest {
assertTrue(result.unmappedKeys.isEmpty());
}
+ // This is a more complicated test which ensures that DeleteRecords requests for multiple
+ // leader nodes are correctly divided up among the nodes based on leadership.
+ // node1 leads t0p0 and t0p2, while node2 leads t0p1 and t0p3.
+ @Test
+ public void testBuildRequestMultipleLeaders() {
+ MetadataResponseData metadataResponseData = new MetadataResponseData();
+ MetadataResponseTopic topicMetadata = new MetadataResponseTopic();
+ topicMetadata.setName("t0").setErrorCode(Errors.NONE.code());
+ topicMetadata.partitions().add(new MetadataResponsePartition()
+ .setPartitionIndex(0).setLeaderId(node1.id()).setErrorCode(Errors.NONE.code()));
+ topicMetadata.partitions().add(new MetadataResponsePartition()
+ .setPartitionIndex(1).setLeaderId(node2.id()).setErrorCode(Errors.NONE.code()));
+ topicMetadata.partitions().add(new MetadataResponsePartition()
+ .setPartitionIndex(2).setLeaderId(node1.id()).setErrorCode(Errors.NONE.code()));
+ topicMetadata.partitions().add(new MetadataResponsePartition()
+ .setPartitionIndex(3).setLeaderId(node2.id()).setErrorCode(Errors.NONE.code()));
+ metadataResponseData.topics().add(topicMetadata);
+ MetadataResponse metadataResponse = new MetadataResponse(metadataResponseData, ApiKeys.METADATA.latestVersion());
+
+ DeleteRecordsHandler handler = new DeleteRecordsHandler(recordsToDelete, logContext, timeout);
+ AdminApiLookupStrategy<TopicPartition> strategy = handler.lookupStrategy();
+ assertInstanceOf(PartitionLeaderStrategy.class, strategy);
+ PartitionLeaderStrategy specificStrategy = (PartitionLeaderStrategy) strategy;
+ MetadataRequest request = specificStrategy.buildRequest(mkSet(t0p0, t0p1, t0p2, t0p3)).build();
+ assertEquals(mkSet("t0"), new HashSet<>(request.topics()));
+
+ Set<TopicPartition> tpSet = mkSet(t0p0, t0p1, t0p2, t0p3);
+ LookupResult<TopicPartition> lookupResult = strategy.handleResponse(tpSet, metadataResponse);
+ assertEquals(emptyMap(), lookupResult.failedKeys);
+ assertEquals(tpSet, lookupResult.mappedKeys.keySet());
+
+ Map<Integer, Set<TopicPartition>> partitionsPerBroker = new HashMap<>();
+ lookupResult.mappedKeys.forEach((tp, node) -> partitionsPerBroker.computeIfAbsent(node, key -> new HashSet<>()).add(tp));
+
+ DeleteRecordsRequest deleteRequest = handler.buildBatchedRequest(node1.id(), partitionsPerBroker.get(node1.id())).build();
+ assertEquals(2, deleteRequest.data().topics().get(0).partitions().size());
+ assertEquals(mkSet(t0p0, t0p2),
+ deleteRequest.data().topics().get(0).partitions().stream()
+ .map(drp -> new TopicPartition("t0", drp.partitionIndex()))
+ .collect(Collectors.toSet()));
+ deleteRequest = handler.buildBatchedRequest(node2.id(), partitionsPerBroker.get(node2.id())).build();
+ assertEquals(2, deleteRequest.data().topics().get(0).partitions().size());
+ assertEquals(mkSet(t0p1, t0p3),
+ deleteRequest.data().topics().get(0).partitions().stream()
+ .map(drp -> new TopicPartition("t0", drp.partitionIndex()))
+ .collect(Collectors.toSet()));
+ }
+
private DeleteRecordsResponse createResponse(Map<TopicPartition, Short> errorsByPartition) {
return createResponse(errorsByPartition, recordsToDelete.keySet());
}
@@ -227,7 +286,7 @@ public class DeleteRecordsHandlerTest {
private AdminApiHandler.ApiResult<TopicPartition, DeletedRecords> handleResponse(DeleteRecordsResponse response) {
DeleteRecordsHandler handler =
new DeleteRecordsHandler(recordsToDelete, logContext, timeout);
- return handler.handleResponse(node, recordsToDelete.keySet(), response);
+ return handler.handleResponse(node1, recordsToDelete.keySet(), response);
}
private void assertResult(