m1a2st commented on code in PR #22815:
URL: https://github.com/apache/kafka/pull/22815#discussion_r3579611277
##########
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);
Review Comment:
`waitForCondition` here doesn't actually wait for anything, the lambda
always return `true`, so it runs exactly once and the 10000 timeout is never
used. It reads as if we're polling, but we aren't.
##########
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java:
##########
@@ -259,6 +260,8 @@ default Set<GroupProtocol> supportedGroupProtocols() {
void startBroker(int brokerId);
+ void restartBroker(int brokerId, Properties propOverrides);
Review Comment:
`Map<String, Object>` should be preferred over `Properties`.
##########
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);
+ }
Review Comment:
This silently turns an empty partitionsToMove into a passing test. we could
add `assertFalse(partitionsToMove.isEmpty());`
--
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]
