mimaison commented on code in PR #22815:
URL: https://github.com/apache/kafka/pull/22815#discussion_r3586137153
##########
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java:
##########
@@ -645,6 +646,50 @@ public void
restartBrokersWithSwappedClientListenerPorts(int nodeId1, int nodeId
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, Properties propOverrides) {
+ BrokerServer broker = brokers.get(nodeId);
+ if (broker == null) {
+ throw new IllegalArgumentException("Unknown broker ID " + nodeId);
+ }
+ if (!broker.isShutdown()) {
+ broker.shutdown();
+ }
Review Comment:
If the current state is `SHUTTING_DOWN` should we call `awaitShutdown()`?
##########
server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java:
##########
@@ -347,6 +350,63 @@ public void testDecommissionBroker() throws
ExecutionException, InterruptedExcep
}
}
+ @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<>();
+ TestUtils.waitForCondition(() -> {
+ LogDirDescription description =
admin.describeLogDirs(List.of(brokerId)).allDescriptions().get().get(brokerId).get(logDirToRemove);
+ partitionsToMove.addAll(description.replicaInfos().keySet());
+ return true;
+ }, 10_000, "Unable to describe log dir " + logDirToRemove);
+ if (!partitionsToMove.isEmpty()) {
+ int target = clusterInstance.brokerIds().stream().filter(id ->
id != brokerId).findFirst().get();
+ movePartitions(admin, partitionsToMove, brokerId,
Optional.of(logDirToRemove), target);
+ }
+
+ // Uncordon log dirs
+ setCordonedLogDirs(admin, List.of(), brokerResource);
+
+ // Physically decommission the log dir: restart the broker without
it in its static log.dirs config
+ Properties propOverrides = new Properties();
+ propOverrides.setProperty(LOG_DIRS_CONFIG, String.join(",",
remainingLogDirs));
+ clusterInstance.restartBroker(brokerId, propOverrides);
+ clusterInstance.waitForReadyBrokers();
Review Comment:
This does not quite follow the decommissioning process:
https://kafka.apache.org/43/operations/basic-kafka-operations/#decommissioning-log-directories
Ideally we'd want to exercise the uncordoning via the controller while the
broker is shut down.
##########
server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java:
##########
@@ -347,6 +350,63 @@ public void testDecommissionBroker() throws
ExecutionException, InterruptedExcep
}
}
+ @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);
Review Comment:
Could we just do `String logDirToRemove = logDirs.remove(logDirs.size() -
1);`, and use `logDirs` in the rest of the test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
