This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 226ce2d8354 KAFKA-20832: Fix cordon logic with user specified
assignments (#22926)
226ce2d8354 is described below
commit 226ce2d8354694625ce98c7213df724f6fead22e
Author: Mickael Maison <[email protected]>
AuthorDate: Mon Jul 27 14:16:02 2026 +0200
KAFKA-20832: Fix cordon logic with user specified assignments (#22926)
Fix the createTopics, createPartitions and alterPartitions code paths
Reviewers: Paolo Patierno <[email protected]>, Gaurav Narula
<[email protected]>, Luke Chen <[email protected]>
---
.../controller/ReplicationControlManager.java | 4 ++
.../controller/ReplicationControlManagerTest.java | 16 +++++-
.../server/CordonedLogDirsIntegrationTest.java | 57 +++++++++++++++++++++-
3 files changed, 74 insertions(+), 3 deletions(-)
diff --git
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
index 4c8a24f0faa..c915615e1ff 100644
---
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -2015,6 +2015,10 @@ public class ReplicationControlManager {
"assignment includes broker " + brokerId + ", but no such
broker is " +
"registered.");
}
+ if
(!clusterControl.brokerRegistrations().get(brokerId).hasUncordonedDirs()) {
+ throw new InvalidReplicaAssignmentException("The manual
partition " +
+ "assignment includes broker " + brokerId + ", but all its
log directories are cordoned.");
+ }
if (brokerId.equals(prevBrokerId)) {
throw new InvalidReplicaAssignmentException("The manual
partition " +
"assignment includes the broker " + prevBrokerId + " more
than " +
diff --git
a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
index b6b66a906e8..71dc3c00cfe 100644
---
a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++
b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -543,6 +543,15 @@ public class ReplicationControlManagerTest {
.collect(Collectors.toSet());
}
+ void cordonBroker(int brokerId) {
+ List<Uuid> dirs =
clusterControl.brokerRegistrations().get(brokerId).directories();
+ BrokerRegistrationChangeRecord record = new
BrokerRegistrationChangeRecord()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(defaultBrokerEpoch(brokerId))
+ .setCordonedLogDirs(dirs);
+ clusterControl.replay(record);
+ }
+
}
static CreateTopicsResponseData withoutConfigs(CreateTopicsResponseData
data) {
@@ -1818,7 +1827,8 @@ public class ReplicationControlManagerTest {
@Test
public void testValidateBadManualPartitionAssignments() {
ReplicationControlTestContext ctx = new
ReplicationControlTestContext.Builder().build();
- ctx.registerBrokers(1, 2);
+ ctx.registerBrokers(0, 1, 2);
+ ctx.cordonBroker(0);
assertEquals("The manual partition assignment includes an empty
replica list.",
assertThrows(InvalidReplicaAssignmentException.class, () ->
ctx.replicationControl.validateManualPartitionAssignment(partitionAssignment(List.of()),
@@ -1827,6 +1837,10 @@ public class ReplicationControlManagerTest {
"broker is registered.",
assertThrows(InvalidReplicaAssignmentException.class, () ->
ctx.replicationControl.validateManualPartitionAssignment(partitionAssignment(List.of(1,
2, 3)),
OptionalInt.empty())).getMessage());
+ assertEquals("The manual partition assignment includes broker 0, but
all its log " +
+ "directories are cordoned.",
assertThrows(InvalidReplicaAssignmentException.class, () ->
+
ctx.replicationControl.validateManualPartitionAssignment(partitionAssignment(List.of(0,
1, 2)),
+ OptionalInt.empty())).getMessage());
assertEquals("The manual partition assignment includes the broker 2
more than " +
"once.", assertThrows(InvalidReplicaAssignmentException.class, ()
->
ctx.replicationControl.validateManualPartitionAssignment(partitionAssignment(List.of(1,
2, 2)),
diff --git
a/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
b/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
index 558c7324d4e..495678d746c 100644
---
a/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
+++
b/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
@@ -27,6 +27,7 @@ import org.apache.kafka.clients.admin.LogDirDescription;
import org.apache.kafka.clients.admin.NewPartitionReassignment;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.UpdateFeaturesOptions;
import org.apache.kafka.clients.admin.UpdateFeaturesResult;
import org.apache.kafka.common.TopicPartition;
@@ -57,6 +58,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import static
org.apache.kafka.server.config.ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -162,18 +164,32 @@ public class CordonedLogDirsIntegrationTest {
setCordonedLogDirs(admin, logDirsBroker0, BROKER_0);
assertCordonedLogDirs(admin, logDirsBroker0);
- // We can't create new topics or partitions
+ // We can't create new topics
Set<NewTopic> newTopics = newTopic(TOPIC2);
Throwable ee = assertThrows(ExecutionException.class, () ->
admin.createTopics(newTopics).all().get()
);
assertInstanceOf(InvalidReplicationFactorException.class,
ee.getCause());
+
+ // We can't create new topics specifying the replica assignment
+ ee = assertThrows(ExecutionException.class, () ->
+ admin.createTopics(Set.of(new NewTopic(TOPIC2, Map.of(0,
List.of(0))))).all().get()
+ );
+ assertInstanceOf(InvalidReplicaAssignmentException.class,
ee.getCause());
+
+ // We can't create new partitions
Map<String, NewPartitions> newPartitions = Map.of(TOPIC1,
NewPartitions.increaseTo(2));
ee = assertThrows(ExecutionException.class, () ->
admin.createPartitions(newPartitions).all().get()
);
assertInstanceOf(InvalidReplicationFactorException.class,
ee.getCause());
+ // We can't create new partitions specifying the replica assignment
+ ee = assertThrows(ExecutionException.class, () ->
+ admin.createPartitions(Map.of(TOPIC1,
NewPartitions.increaseTo(2, List.of(List.of(0))))).all().get()
+ );
+ assertInstanceOf(InvalidReplicaAssignmentException.class,
ee.getCause());
+
// Uncordon all log dirs
setCordonedLogDirs(admin, List.of(logDirsBroker0.get(0)),
BROKER_0);
assertCordonedLogDirs(admin, List.of(logDirsBroker0.get(0)));
@@ -207,7 +223,7 @@ public class CordonedLogDirsIntegrationTest {
}
@ClusterTest
- public void testReassignWithCordonedLogDirs() throws Exception {
+ public void testAlterReplicaWithCordonedLogDirs() throws Exception {
TopicPartitionReplica replica = new TopicPartitionReplica(TOPIC1, 0,
0);
try (Admin admin = clusterInstance.admin()) {
admin.createTopics(newTopic(TOPIC1)).all().get();
@@ -235,6 +251,43 @@ public class CordonedLogDirsIntegrationTest {
}
}
+ @ClusterTest(
+ brokers = 2
+ )
+ public void testAlterPartitionWithCordonedLogDirs() throws Exception {
+ Set<Integer> allBrokers = new HashSet<>(clusterInstance.brokerIds());
+ try (Admin admin = clusterInstance.admin()) {
+ admin.createTopics(newTopic(TOPIC1)).all().get();
+
+ // Find the broker that hosts the partition and cordon the other
broker
+ AtomicReference<Integer> partitionBroker = new AtomicReference<>();
+ TestUtils.waitForCondition(() -> {
+ TopicDescription td =
admin.describeTopics(List.of(TOPIC1)).allTopicNames().get().get(TOPIC1);
+ if (td == null) return false;
+
partitionBroker.set(td.partitions().get(0).replicas().get(0).id());
+ return true;
+ }, 10_000, "Unable to find broker hosting topic " + TOPIC1);
+ allBrokers.remove(partitionBroker.get());
+ assertEquals(1, allBrokers.size());
+ int otherBroker = allBrokers.iterator().next();
+ setCordonedLogDirs(admin, List.of("*"), new
ConfigResource(ConfigResource.Type.BROKER, String.valueOf(otherBroker)));
+
+ // We can't move a partition to the other broker
+ Throwable ee = assertThrows(ExecutionException.class, () ->
+ admin.alterPartitionReassignments(Map.of(
+ new TopicPartition(TOPIC1, 0),
+ Optional.of(new
NewPartitionReassignment(List.of(otherBroker))))).all().get()
+ );
+ assertInstanceOf(InvalidReplicaAssignmentException.class,
ee.getCause());
+
+ // After uncordoning the other broker, we can move the replica on
it
+ setCordonedLogDirs(admin, List.of(), new
ConfigResource(ConfigResource.Type.BROKER, String.valueOf(otherBroker)));
+ admin.alterPartitionReassignments(Map.of(
+ new TopicPartition(TOPIC1, 0),
+ Optional.of(new
NewPartitionReassignment(List.of(otherBroker))))).all().get();
+ }
+ }
+
@ClusterTest()
public void testCordonUnknownLogDirs() {
try (Admin admin = clusterInstance.admin()) {
