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 2026/08/03 16:12:33 UTC
(kafka) branch trunk updated: KAFKA-20215: Add integration test for log directory cordoning (#22815)
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 ff1f4674063 KAFKA-20215: Add integration test for log directory cordoning (#22815)
ff1f4674063 is described below
commit ff1f46740636a32d7d2155ea84c410b92ca07318
Author: Chang-Chi Hsu <ji...@gmail.com>
AuthorDate: Tue Aug 4 00:12:26 2026 +0800
KAFKA-20215: Add integration test for log directory cordoning (#22815)
- Added an integration test for log directory decommissioning.
- The test verifies that a broker can restart successfully after a log
directory has been decommissioned.
Reviewers: Ken Huang <s7...@gmail.com>, Mickael Maison <mi...@gmail.com>
---
.../server/CordonedLogDirsIntegrationTest.java | 61 ++++++++++++++++++++++
.../apache/kafka/common/test/ClusterInstance.java | 2 +
.../kafka/common/test/KafkaClusterTestKit.java | 44 ++++++++++++++++
.../test/junit/RaftClusterInvocationContext.java | 5 ++
4 files changed, 112 insertions(+)
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 495678d746c..c7a67051496 100644
--- a/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
+++ b/server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java
@@ -58,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.apache.kafka.server.config.ServerLogConfigs.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;
@@ -400,6 +401,66 @@ public class CordonedLogDirsIntegrationTest {
}
}
+ @ClusterTest(
+ brokers = 2,
+ controllers = 1
+ )
+ public void testDecommissionLogDir() throws ExecutionException, InterruptedException {
+ // Select the target broker
+ int brokerId = clusterInstance.brokerIds().stream().filter(id -> !clusterInstance.controllerIds().contains(id)).findFirst().get();
+ try (var admin = clusterInstance.admin()) {
+ List<String> logDirs = clusterInstance.brokers().get(brokerId).config().logDirs();
+ assertTrue(logDirs.size() > 1, "Test requires more than one log dir per broker");
+ String logDirToRemove = logDirs.get(logDirs.size() - 1);
+ List<String> remainingLogDirs = logDirs.subList(0, logDirs.size() - 1);
+
+ // Create 10 topics, replicated to every broker so brokerId is guaranteed to host some
+ // replicas on the log dir we're about to decommission
+ for (int i = 0; i < 10; i++) {
+ admin.createTopics(newTopic("topic" + i)).all().get();
+ }
+ TestUtils.waitForCondition(() -> admin.listTopics().names().get().size() == 10, 10_000, "Topics were not created");
+
+ ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(brokerId));
+
+ // Cordon the log dir we're going to decommission and move any replicas hosted on it elsewhere
+ setCordonedLogDirs(admin, List.of(logDirToRemove), brokerResource);
+ Set<TopicPartition> partitionsToMove = new HashSet<>();
+ LogDirDescription description = admin.describeLogDirs(List.of(brokerId)).allDescriptions().get().get(brokerId).get(logDirToRemove);
+ partitionsToMove.addAll(description.replicaInfos().keySet());
+ assertFalse(partitionsToMove.isEmpty());
+ int target = clusterInstance.brokerIds().stream().filter(id -> id != brokerId).findFirst().get();
+ movePartitions(admin, partitionsToMove, brokerId, Optional.of(logDirToRemove), target);
+
+ // Shut down the broker so the log dir can be physically decommissioned while it's offline
+ clusterInstance.brokers().get(brokerId).shutdown();
+ clusterInstance.brokers().get(brokerId).awaitShutdown();
+
+ // Uncordon the log dir via the controller while the broker is shut down
+ try (Admin controllerAdmin = clusterInstance.admin(Map.of(), true)) {
+ controllerAdmin.incrementalAlterConfigs(cordonedDirsConfig("", brokerResource)).all().get();
+ }
+
+ // Physically decommission the log dir: restart the
+ // broker without it in its static log.dirs config
+ Map<String, Object> propOverrides = Map.of(LOG_DIRS_CONFIG, String.join(",", remainingLogDirs), CORDONED_LOG_DIRS_CONFIG, "");
+ clusterInstance.restartBroker(brokerId, propOverrides);
+ clusterInstance.waitForReadyBrokers();
+
+ // The broker restarted with the reduced set of log dirs, and no longer reports the removed one
+ assertEquals(remainingLogDirs, clusterInstance.brokers().get(brokerId).config().logDirs());
+ TestUtils.waitForCondition(() ->
+ !admin.describeLogDirs(List.of(brokerId)).allDescriptions().get().get(brokerId).containsKey(logDirToRemove),
+ 10_000, "Broker " + brokerId + " is still reporting the removed log dir " + logDirToRemove);
+
+ // The broker is still fully functional after losing a log dir
+ admin.createTopics(newTopic("topic-after-decommission")).all().get();
+ TestUtils.waitForCondition(() ->
+ admin.listTopics().names().get().size() == 11,
+ 10_000, "Topic was not created after decommissioning a log dir");
+ }
+ }
+
private void movePartitions(Admin admin, Set<TopicPartition> partitions, int source, Optional<String> logDir, int target) throws ExecutionException, InterruptedException {
Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>();
for (TopicPartition partition : partitions) {
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
index cdf911f5198..dacf3f27241 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
@@ -259,6 +259,8 @@ public interface ClusterInstance {
void startBroker(int brokerId);
+ void restartBroker(int brokerId, Map<String, Object> propOverrides);
+
void restartBrokersWithSwappedClientListenerPorts(int brokerId1, int brokerId2);
//---------------------------[wait]---------------------------//
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
index eec40e2bb82..ec2fa992d9c 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
@@ -647,6 +647,50 @@ public class KafkaClusterTestKit implements AutoCloseable {
broker2.startup();
}
+ /**
+ * Shuts down the given broker (if it isn't already) and starts it back up with a possibly
+ * modified static configuration. This allows tests to change read-only configs, such as
+ * {@code log.dirs}, which can only be applied when the broker process (re)starts.
+ * <p>
+ * The broker keeps its identity (node ID, cluster metadata, bound ports): a new
+ * {@link SharedServer}/{@link BrokerServer} pair is created from the previous broker's
+ * {@link MetaPropertiesEnsemble} and socket factory, but with a {@link KafkaConfig} derived
+ * from the previous one with {@code propOverrides} applied on top.
+ *
+ * @param nodeId The ID of the broker to restart.
+ * @param propOverrides Configs to override in the broker's static configuration.
+ */
+ public void restartBroker(int nodeId, Map<String, Object> propOverrides) {
+ BrokerServer broker = brokers.get(nodeId);
+ if (broker == null) {
+ throw new IllegalArgumentException("Unknown broker ID " + nodeId);
+ }
+ if (!broker.isShutdown()) {
+ broker.shutdown();
+ }
+ broker.awaitShutdown();
+
+ Map<String, Object> props = new HashMap<>(broker.config().originals());
+ props.putAll(propOverrides);
+ KafkaConfig newConfig = new KafkaConfig(props, false);
+
+ SharedServer sharedServer = new SharedServer(
+ newConfig,
+ broker.sharedServer().metaPropsEnsemble(),
+ Time.SYSTEM,
+ new Metrics(),
+ CompletableFuture.completedFuture(
+ QuorumConfig.parseVoterConnections(newConfig.quorumConfig().voters())),
+ QuorumConfig.parseBootstrapServers(newConfig.quorumConfig().bootstrapServers()),
+ faultHandlerFactory,
+ socketFactoryManager.getOrCreateSocketFactory(nodeId)
+ );
+ broker = new BrokerServer(sharedServer);
+ brokers.put(nodeId, broker);
+
+ broker.startup();
+ }
+
/**
* Wait for a controller to mark all the brokers as ready (registered and unfenced).
* And also wait for the metadata cache up-to-date in each broker server.
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
index 699adaa3ec7..75757b9f4e1 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java
@@ -271,6 +271,11 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
findBrokerOrThrow(brokerId).startup();
}
+ @Override
+ public void restartBroker(int brokerId, Map<String, Object> propOverrides) {
+ clusterTestKit.restartBroker(brokerId, propOverrides);
+ }
+
@Override
public void restartBrokersWithSwappedClientListenerPorts(int brokerId1, int brokerId2) {
try {