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 <[email protected]>
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 <[email protected]>, Mickael Maison
<[email protected]>
---
.../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 {
