You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by so...@apache.org on 2024/06/08 13:13:03 UTC
(kafka) branch 3.8 updated: KAFKA-16886: Detect replica demotion in AssignmentsManager (#16232)
This is an automated email from the ASF dual-hosted git repository.
soarez pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new d5c9381e06b KAFKA-16886: Detect replica demotion in AssignmentsManager (#16232)
d5c9381e06b is described below
commit d5c9381e06b226603e11c0bf5de5dd25e5689fa3
Author: Igor Soarez <i...@soarez.me>
AuthorDate: Sat Jun 8 16:10:23 2024 +0300
KAFKA-16886: Detect replica demotion in AssignmentsManager (#16232)
JBOD Brokers keep the Controller up to date with replica-to-directory
placement via AssignReplicasToDirsRequest. These requests are queued,
compacted and sent by AssignmentsManager.
The Controller returns the error NOT_LEADER_OR_FOLLOWER when handling
a AssignReplicasToDirsRequest from a broker that is not a replica.
A partition reassignment can take place, removing the Broker
as a replica before the AssignReplicasToDirsRequest successfully
reaches the Controller. AssignmentsManager retries failed
requests, and will continuously try to propagate this assignment,
until the Broker either shuts down, or is added back as a replica.
When encountering a NOT_LEADER_OR_FOLLOWER error, AssignmentsManager
should assume that the broker is no longer a replica, and stop
trying to propagate the directory assignment for that partition.
Reviewers: Luke Chen <sh...@gmail.com>
---
.../apache/kafka/server/AssignmentsManager.java | 8 ++-
.../kafka/server/AssignmentsManagerTest.java | 58 +++++++++++++++++++++-
2 files changed, 64 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java b/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
index bcdfd13cf72..4a00b023395 100644
--- a/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
+++ b/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
@@ -128,6 +128,10 @@ public class AssignmentsManager {
}
}
+ public void onAssignment(TopicIdPartition topicPartition, Uuid dirId, String reason) {
+ onAssignment(topicPartition, dirId, reason, null);
+ }
+
public void onAssignment(TopicIdPartition topicPartition, Uuid dirId, String reason, Runnable callback) {
if (callback == null) {
callback = () -> { };
@@ -444,7 +448,9 @@ public class AssignmentsManager {
} else {
acknowledged.add(topicPartition);
Errors error = Errors.forCode(partition.errorCode());
- if (error != Errors.NONE) {
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER) {
+ log.info("Dropping late directory assignment for partition {} into directory {} because this broker is no longer a replica", partition, event.dirId);
+ } else if (error != Errors.NONE) {
log.error("Controller returned error {} for assignment of partition {} into directory {}",
error.name(), partition, event.dirId);
failures.add(event);
diff --git a/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java b/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
index a385e2d8c9c..457f5bd59a9 100644
--- a/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
+++ b/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
@@ -40,15 +40,18 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.mockito.ArgumentCaptor;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import java.util.function.Supplier;
import static org.apache.kafka.metadata.AssignmentsHelper.buildRequestData;
@@ -310,12 +313,21 @@ public class AssignmentsManagerTest {
}
private static ClientResponse buildSuccessfulResponse(AssignReplicasToDirsRequestData request) {
+ return buildResponse(request, topicIdPartition -> Errors.NONE);
+ }
+
+ private static ClientResponse buildResponse(AssignReplicasToDirsRequestData request,
+ Function<TopicIdPartition, Errors> perPartitionError) {
Map<Uuid, Map<TopicIdPartition, Errors>> errors = new HashMap<>();
for (AssignReplicasToDirsRequestData.DirectoryData directory : request.directories()) {
for (AssignReplicasToDirsRequestData.TopicData topic : directory.topics()) {
for (AssignReplicasToDirsRequestData.PartitionData partition : topic.partitions()) {
TopicIdPartition topicIdPartition = new TopicIdPartition(topic.topicId(), partition.partitionIndex());
- errors.computeIfAbsent(directory.id(), d -> new HashMap<>()).put(topicIdPartition, Errors.NONE);
+ Errors error = perPartitionError.apply(topicIdPartition);
+ if (error == null) {
+ error = Errors.NONE;
+ }
+ errors.computeIfAbsent(directory.id(), d -> new HashMap<>()).put(topicIdPartition, error);
}
}
}
@@ -403,4 +415,48 @@ public class AssignmentsManagerTest {
}
TestUtils.retryOnExceptionWithTimeout(5_000, () -> assertEquals(8, queuedReplicaToDirAssignments.value()));
}
+
+ // AssignmentsManager retries to propagate assignments (via AssignReplicasToDirsRequest) after failures.
+ // When an assignment fails to propagate with NOT_LEADER_OR_FOLLOWER, AssignmentsManager should conclude
+ // that the broker has been removed as a replica for the partition, and stop trying to propagate it.
+ @Test
+ void testDropsOldAssignments() throws InterruptedException {
+ TopicIdPartition tp1 = new TopicIdPartition(TOPIC_1, 1), tp2 = new TopicIdPartition(TOPIC_1, 2);
+ List<AssignReplicasToDirsRequestData> requests = new ArrayList<>();
+ CountDownLatch readyToAssert = new CountDownLatch(2);
+ doAnswer(invocation -> {
+ AssignReplicasToDirsRequestData request = invocation.getArgument(0, AssignReplicasToDirsRequest.Builder.class).build().data();
+ ControllerRequestCompletionHandler completionHandler = invocation.getArgument(1, ControllerRequestCompletionHandler.class);
+ if (readyToAssert.getCount() == 2) {
+ // First request, reply with a partition-level NOT_LEADER_OR_FOLLOWER error and queue a different assignment
+ completionHandler.onComplete(buildResponse(request, topicIdPartition -> Errors.NOT_LEADER_OR_FOLLOWER));
+ manager.onAssignment(tp2, DIR_1, "testDropsOldAssignments-second");
+ }
+ if (readyToAssert.getCount() == 1) {
+ // Second request, reply with success
+ completionHandler.onComplete(buildSuccessfulResponse(request));
+ }
+ requests.add(request);
+ readyToAssert.countDown();
+ return null;
+ }).when(channelManager).sendRequest(any(), any());
+
+ manager.onAssignment(tp1, DIR_1, "testDropsOldAssignments-first");
+ TestUtils.waitForCondition(() -> {
+ time.sleep(TimeUnit.SECONDS.toMillis(1));
+ manager.wakeup();
+ return readyToAssert.await(1, TimeUnit.MILLISECONDS);
+ }, "Timed out waiting for AssignReplicasToDirsRequest to be sent.");
+
+ assertEquals(Arrays.asList(
+ buildRequestData(8, 100, new HashMap<TopicIdPartition, Uuid>() {{
+ put(tp1, DIR_1);
+ }}),
+ // Even though the controller replied with NOT_LEADER_OR_FOLLOWER, the second request does not include
+ // partition 1, meaning AssignmentManager dropped (no longer retries) the assignment.
+ buildRequestData(8, 100, new HashMap<TopicIdPartition, Uuid>() {{
+ put(tp2, DIR_1);
+ }})
+ ), requests);
+ }
}