You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2025/08/07 10:20:36 UTC
[PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
m1a2st opened a new pull request, #20319:
URL: https://github.com/apache/kafka/pull/20319
The ReassignPartitionsCommand shows the topic replicas on each broker. When using the --generate command, it returns the current partition replica assignment. However, the log directory for each current replica is always shown as any. This makes it impossible for users to determine which specific log directory is being used by each replica. Therefore, we should fix this behavior.
```
Current partition replica assignment
{"version":1,"partitions":[{"topic":"test1","partition":0,"replicas":[4,2],"log_dirs":["any","any"]},{"topic":"test1","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"test1","partition":2,"replicas":[3,4],"log_dirs":["any","any"]},{"topic":"test1","partition":3,"replicas":[2,4],"log_dirs":["any","any"]},{"topic":"test1","partition":4,"replicas":[4,3],"log_dirs":["any","any"]},{"topic":"test1","partition":5,"replicas":[3,2],"log_dirs":["any","any"]},{"topic":"test2","partition":0,"replicas":[4,2],"log_dirs":["any","any"]},{"topic":"test2","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"test2","partition":2,"replicas":[3,4],"log_dirs":["any","any"]}]}
```
This PR
```
Current partition replica assignment
{"version":1,"partitions":[{"topic":"test1","partition":0,"replicas":[4,2],"log_dirs":["/tmp/kraft-broker-logs234","/tmp/kraft-broker-logs"]},{"topic":"test1","partition":1,"replicas":[2,3],"log_dirs":["/tmp/kraft-broker-logs","/tmp/kraft-broker-logs123"]},{"topic":"test1","partition":2,"replicas":[3,4],"log_dirs":["/tmp/kraft-broker-logs123","/tmp/kraft-broker-logs234"]},{"topic":"test1","partition":3,"replicas":[2,4],"log_dirs":["/tmp/kraft-broker-logs","/tmp/kraft-broker-logs234"]},{"topic":"test1","partition":4,"replicas":[4,3],"log_dirs":["/tmp/kraft-broker-logs234","/tmp/kraft-broker-logs123"]},{"topic":"test1","partition":5,"replicas":[3,2],"log_dirs":["/tmp/kraft-broker-logs123","/tmp/kraft-broker-logs"]},{"topic":"test2","partition":0,"replicas":[4,2],"log_dirs":["/tmp/kraft-broker-logs234","/tmp/kraft-broker-logs"]},{"topic":"test2","partition":1,"replicas":[2,3],"log_dirs":["/tmp/kraft-broker-logs","/tmp/kraft-broker-logs123"]},{"topic":"test2","partition":2,"replicas":
[3,4],"log_dirs":["/tmp/kraft-broker-logs123","/tmp/kraft-broker-logs234"]}]}
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2291766406
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1514,4 +1520,26 @@ static Set<TopicPartitionReplica> alterReplicaLogDirs(Admin adminClient,
}
return results;
}
+
+ static Map<TopicPartitionReplica, String> getReplicaToLogDir(Admin adminClient, List<TopicPartitionReplica> replicas) throws InterruptedException, ExecutionException {
+ return adminClient.describeReplicaLogDirs(replicas).all().get()
+ .entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() != null && entry.getValue().getCurrentReplicaLogDir() != null)
Review Comment:
why to add `entry.getValue() != null` check?
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1514,4 +1520,26 @@ static Set<TopicPartitionReplica> alterReplicaLogDirs(Admin adminClient,
}
return results;
}
+
+ static Map<TopicPartitionReplica, String> getReplicaToLogDir(Admin adminClient, List<TopicPartitionReplica> replicas) throws InterruptedException, ExecutionException {
+ return adminClient.describeReplicaLogDirs(replicas).all().get()
+ .entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() != null && entry.getValue().getCurrentReplicaLogDir() != null)
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().getCurrentReplicaLogDir()
+ ));
+ }
+
+ private static List<TopicPartitionReplica> getTopicPartitionReplicas(Map<TopicPartition, List<Integer>> topicPartitionToReplicas) {
Review Comment:
It seems `getTopicPartitionReplicas` and `getReplicaToLogDir` can be merged into a single helper.
```java
static Map<TopicPartitionReplica, String> getReplicaToLogDir(Admin adminClient,
Map<TopicPartition, List<Integer>> topicPartitionToReplicas) throws InterruptedException, ExecutionException {
var tprs = topicPartitionToReplicas.entrySet().stream()
.flatMap(entry -> entry.getValue().stream()
.map(id -> new TopicPartitionReplica(entry.getKey().topic(), entry.getKey().partition(), id)))
.collect(Collectors.toUnmodifiableSet());
return adminClient.describeReplicaLogDirs(tprs).all().get()
.entrySet()
.stream()
.filter(entry -> entry.getValue().getCurrentReplicaLogDir() != null)
.collect(Collectors.toMap(
Entry::getKey,
entry -> entry.getValue().getCurrentReplicaLogDir()
));
}
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2265554451
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1514,4 +1520,30 @@ static Set<TopicPartitionReplica> alterReplicaLogDirs(Admin adminClient,
}
return results;
}
+
+ static Map<TopicPartitionReplica, String> getReplicaToLogDir(Admin adminClient, List<TopicPartitionReplica> replicas) {
+ try {
+ return adminClient.describeReplicaLogDirs(replicas).all().get()
+ .entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() != null && entry.getValue().getCurrentReplicaLogDir() != null)
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().getCurrentReplicaLogDir()
+ ));
+ } catch (InterruptedException | ExecutionException e) {
+ throw new AdminOperationException(e.getCause());
+ }
Review Comment:
We don't do other handling for `InterruptedException` and `ExecutionException`. Could we move the exceptions to function signatures?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2263576388
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -436,29 +437,50 @@ public void testGenerateAssignmentWithFewerBrokers() throws Exception {
@Test
public void testCurrentPartitionReplicaAssignmentToString() throws Exception {
- Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
-
- proposedParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
- proposedParts.put(new TopicPartition("bar", 0), List.of(7, 8, 9));
-
- Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
-
- currentParts.put(new TopicPartition("foo", 0), List.of(1, 2, 3));
- currentParts.put(new TopicPartition("foo", 1), List.of(4, 5, 6));
- currentParts.put(new TopicPartition("bar", 0), List.of(7, 8));
- currentParts.put(new TopicPartition("baz", 0), List.of(10, 11, 12));
+ try (MockAdminClient adminClient = new MockAdminClient.Builder()
+ .numBrokers(6)
+ .brokerLogDirs(List.of(
+ List.of("/tmp/broker0/logs"),
+ List.of("/tmp/broker1/logs"),
+ List.of("/tmp/broker2/logs"),
+ List.of("/tmp/broker3/logs"),
+ List.of("/tmp/broker4/logs"),
+ List.of("/tmp/broker5/logs")
+ ))
+ .build()
+ ) {
+
+ List<Node> brokers = adminClient.brokers();
+ adminClient.addTopic(false, "foo", List.of(
+ new TopicPartitionInfo(1, brokers.get(1),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)))
+ ), Map.of());
+
+ adminClient.addTopic(false, "bar", List.of(
+ new TopicPartitionInfo(0, brokers.get(4),
+ List.of(brokers.get(4), brokers.get(5)),
+ List.of(brokers.get(4), brokers.get(5)))
+ ), Map.of());
+
+ Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
+ proposedParts.put(new TopicPartition("foo", 1), List.of(0, 1, 2));
+ proposedParts.put(new TopicPartition("bar", 0), List.of(3, 4, 5));
+
+ Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
+ currentParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
+ currentParts.put(new TopicPartition("bar", 0), List.of(4, 5));
- assertEquals(String.join(System.lineSeparator(),
- "Current partition replica assignment",
- "",
- "{\"version\":1,\"partitions\":" +
- "[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[7,8],\"log_dirs\":[\"any\",\"any\"]}," +
- "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[4,5,6],\"log_dirs\":[\"any\",\"any\",\"any\"]}]" +
- "}",
- "",
- "Save this to use as the --reassignment-json-file option during rollback"),
- currentPartitionReplicaAssignmentToString(proposedParts, currentParts)
- );
+ assertEquals(String.join(System.lineSeparator(),
+ "Current partition replica assignment",
+ "",
+ "{\"version\":1,\"partitions\":[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[4,5],\"log_dirs\":[\"/tmp/broker4/logs\",\"/tmp/broker4/logs\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}]}",
Review Comment:
'any' looks like a bug. I believe this is caused by this line:
https://github.com/apache/kafka/blob/8deb6c6911616f887ebb2678f3f12ee1da09a618/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java#L1135
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2264166649
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -436,29 +437,50 @@ public void testGenerateAssignmentWithFewerBrokers() throws Exception {
@Test
public void testCurrentPartitionReplicaAssignmentToString() throws Exception {
- Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
-
- proposedParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
- proposedParts.put(new TopicPartition("bar", 0), List.of(7, 8, 9));
-
- Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
-
- currentParts.put(new TopicPartition("foo", 0), List.of(1, 2, 3));
- currentParts.put(new TopicPartition("foo", 1), List.of(4, 5, 6));
- currentParts.put(new TopicPartition("bar", 0), List.of(7, 8));
- currentParts.put(new TopicPartition("baz", 0), List.of(10, 11, 12));
+ try (MockAdminClient adminClient = new MockAdminClient.Builder()
+ .numBrokers(6)
+ .brokerLogDirs(List.of(
+ List.of("/tmp/broker0/logs"),
+ List.of("/tmp/broker1/logs"),
+ List.of("/tmp/broker2/logs"),
+ List.of("/tmp/broker3/logs"),
+ List.of("/tmp/broker4/logs"),
+ List.of("/tmp/broker5/logs")
+ ))
+ .build()
+ ) {
+
+ List<Node> brokers = adminClient.brokers();
+ adminClient.addTopic(false, "foo", List.of(
+ new TopicPartitionInfo(1, brokers.get(1),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)))
+ ), Map.of());
+
+ adminClient.addTopic(false, "bar", List.of(
+ new TopicPartitionInfo(0, brokers.get(4),
+ List.of(brokers.get(4), brokers.get(5)),
+ List.of(brokers.get(4), brokers.get(5)))
+ ), Map.of());
+
+ Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
+ proposedParts.put(new TopicPartition("foo", 1), List.of(0, 1, 2));
+ proposedParts.put(new TopicPartition("bar", 0), List.of(3, 4, 5));
+
+ Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
+ currentParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
+ currentParts.put(new TopicPartition("bar", 0), List.of(4, 5));
- assertEquals(String.join(System.lineSeparator(),
- "Current partition replica assignment",
- "",
- "{\"version\":1,\"partitions\":" +
- "[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[7,8],\"log_dirs\":[\"any\",\"any\"]}," +
- "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[4,5,6],\"log_dirs\":[\"any\",\"any\",\"any\"]}]" +
- "}",
- "",
- "Save this to use as the --reassignment-json-file option during rollback"),
- currentPartitionReplicaAssignmentToString(proposedParts, currentParts)
- );
+ assertEquals(String.join(System.lineSeparator(),
+ "Current partition replica assignment",
+ "",
+ "{\"version\":1,\"partitions\":[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[4,5],\"log_dirs\":[\"/tmp/broker4/logs\",\"/tmp/broker4/logs\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}]}",
Review Comment:
However, changing this behavior would affect other tests, so I believe we should not modify it in this PR.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2262947299
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -922,14 +924,17 @@ private static void verifyBrokerIds(Admin adminClient, Set<Integer> brokers) thr
* @return The string to print. We will only print information about
* partitions that appear in the proposed partition assignment.
*/
- static String currentPartitionReplicaAssignmentToString(Map<TopicPartition, List<Integer>> proposedParts,
+ static String currentPartitionReplicaAssignmentToString(Admin adminClient,
Review Comment:
Nit: Please also update the param in Java docs.
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -436,29 +437,50 @@ public void testGenerateAssignmentWithFewerBrokers() throws Exception {
@Test
public void testCurrentPartitionReplicaAssignmentToString() throws Exception {
- Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
-
- proposedParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
- proposedParts.put(new TopicPartition("bar", 0), List.of(7, 8, 9));
-
- Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
-
- currentParts.put(new TopicPartition("foo", 0), List.of(1, 2, 3));
- currentParts.put(new TopicPartition("foo", 1), List.of(4, 5, 6));
- currentParts.put(new TopicPartition("bar", 0), List.of(7, 8));
- currentParts.put(new TopicPartition("baz", 0), List.of(10, 11, 12));
+ try (MockAdminClient adminClient = new MockAdminClient.Builder()
+ .numBrokers(6)
+ .brokerLogDirs(List.of(
+ List.of("/tmp/broker0/logs"),
+ List.of("/tmp/broker1/logs"),
+ List.of("/tmp/broker2/logs"),
+ List.of("/tmp/broker3/logs"),
+ List.of("/tmp/broker4/logs"),
+ List.of("/tmp/broker5/logs")
+ ))
+ .build()
+ ) {
+
+ List<Node> brokers = adminClient.brokers();
+ adminClient.addTopic(false, "foo", List.of(
+ new TopicPartitionInfo(1, brokers.get(1),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)))
+ ), Map.of());
+
+ adminClient.addTopic(false, "bar", List.of(
+ new TopicPartitionInfo(0, brokers.get(4),
+ List.of(brokers.get(4), brokers.get(5)),
+ List.of(brokers.get(4), brokers.get(5)))
+ ), Map.of());
+
+ Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
+ proposedParts.put(new TopicPartition("foo", 1), List.of(0, 1, 2));
+ proposedParts.put(new TopicPartition("bar", 0), List.of(3, 4, 5));
+
+ Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
+ currentParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
+ currentParts.put(new TopicPartition("bar", 0), List.of(4, 5));
- assertEquals(String.join(System.lineSeparator(),
- "Current partition replica assignment",
- "",
- "{\"version\":1,\"partitions\":" +
- "[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[7,8],\"log_dirs\":[\"any\",\"any\"]}," +
- "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[4,5,6],\"log_dirs\":[\"any\",\"any\",\"any\"]}]" +
- "}",
- "",
- "Save this to use as the --reassignment-json-file option during rollback"),
- currentPartitionReplicaAssignmentToString(proposedParts, currentParts)
- );
+ assertEquals(String.join(System.lineSeparator(),
+ "Current partition replica assignment",
+ "",
+ "{\"version\":1,\"partitions\":[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[4,5],\"log_dirs\":[\"/tmp/broker4/logs\",\"/tmp/broker4/logs\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}]}",
Review Comment:
'any' looks like a bug. I believe this is caused by this line:
https://github.com/apache/kafka/blob/8deb6c6911616f887ebb2678f3f12ee1da09a618/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java#L1135
Is bug i think
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "andrewgrantcflt (via GitHub)" <gi...@apache.org>.
andrewgrantcflt commented on PR #20319:
URL: https://github.com/apache/kafka/pull/20319#issuecomment-3696928198
> @FrankYang0529 @chia7712 it seems with this change, you now can't use `kafka-reassign-partitions.sh ... --execute` to reassign partitions away from dead brokers i.e. brokers that can't be reached. I ran into this in an environment of mine. I had some dead brokers and when I tried to reassign partitions away from them I got
>
> ```
> Error: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeReplicaLogDirs
> ...
> ```
>
> Seems problematic?
I created https://issues.apache.org/jira/browse/KAFKA-20023 to address this.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2264592086
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -436,29 +437,50 @@ public void testGenerateAssignmentWithFewerBrokers() throws Exception {
@Test
public void testCurrentPartitionReplicaAssignmentToString() throws Exception {
- Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
-
- proposedParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
- proposedParts.put(new TopicPartition("bar", 0), List.of(7, 8, 9));
-
- Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
-
- currentParts.put(new TopicPartition("foo", 0), List.of(1, 2, 3));
- currentParts.put(new TopicPartition("foo", 1), List.of(4, 5, 6));
- currentParts.put(new TopicPartition("bar", 0), List.of(7, 8));
- currentParts.put(new TopicPartition("baz", 0), List.of(10, 11, 12));
+ try (MockAdminClient adminClient = new MockAdminClient.Builder()
+ .numBrokers(6)
+ .brokerLogDirs(List.of(
+ List.of("/tmp/broker0/logs"),
+ List.of("/tmp/broker1/logs"),
+ List.of("/tmp/broker2/logs"),
+ List.of("/tmp/broker3/logs"),
+ List.of("/tmp/broker4/logs"),
+ List.of("/tmp/broker5/logs")
+ ))
+ .build()
+ ) {
+
+ List<Node> brokers = adminClient.brokers();
+ adminClient.addTopic(false, "foo", List.of(
+ new TopicPartitionInfo(1, brokers.get(1),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)))
+ ), Map.of());
+
+ adminClient.addTopic(false, "bar", List.of(
+ new TopicPartitionInfo(0, brokers.get(4),
+ List.of(brokers.get(4), brokers.get(5)),
+ List.of(brokers.get(4), brokers.get(5)))
+ ), Map.of());
+
+ Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
+ proposedParts.put(new TopicPartition("foo", 1), List.of(0, 1, 2));
+ proposedParts.put(new TopicPartition("bar", 0), List.of(3, 4, 5));
+
+ Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
+ currentParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
+ currentParts.put(new TopicPartition("bar", 0), List.of(4, 5));
- assertEquals(String.join(System.lineSeparator(),
- "Current partition replica assignment",
- "",
- "{\"version\":1,\"partitions\":" +
- "[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[7,8],\"log_dirs\":[\"any\",\"any\"]}," +
- "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[4,5,6],\"log_dirs\":[\"any\",\"any\",\"any\"]}]" +
- "}",
- "",
- "Save this to use as the --reassignment-json-file option during rollback"),
- currentPartitionReplicaAssignmentToString(proposedParts, currentParts)
- );
+ assertEquals(String.join(System.lineSeparator(),
+ "Current partition replica assignment",
+ "",
+ "{\"version\":1,\"partitions\":[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[4,5],\"log_dirs\":[\"/tmp/broker4/logs\",\"/tmp/broker4/logs\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}]}",
Review Comment:
It's fine for me right now. I'll track this. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20319:
URL: https://github.com/apache/kafka/pull/20319
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "andrewgrantcflt (via GitHub)" <gi...@apache.org>.
andrewgrantcflt commented on PR #20319:
URL: https://github.com/apache/kafka/pull/20319#issuecomment-3696871283
@FrankYang0529 @chia7712 it seems with this change, you now can't use `kafka-reassign-partitions.sh ... --execute` to reassign partitions away from dead brokers i.e. brokers that can't be reached. I ran into this in an environment of mine. I had some dead brokers and when I tried to reassign partitions away from them I got
```
Error: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeReplicaLogDirs
...
```
Seems problematic?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2294450811
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -566,10 +566,11 @@ public static Entry<Map<TopicPartition, List<Integer>>, Map<TopicPartition, List
List<String> topicsToReassign = t0.getValue();
Map<TopicPartition, List<Integer>> currentAssignments = getReplicaAssignmentForTopics(adminClient, topicsToReassign);
+ Map<TopicPartitionReplica, String> currentReplicaLogDirs = getReplicaToLogDir(adminClient, currentAssignments);
List<UsableBroker> usableBrokers = getBrokerMetadata(adminClient, brokersToReassign, enableRackAwareness);
Map<TopicPartition, List<Integer>> proposedAssignments = calculateAssignment(currentAssignments, usableBrokers);
System.out.printf("Current partition replica assignment%n%s%n%n",
- formatAsReassignmentJson(currentAssignments, Map.of()));
+ formatAsReassignmentJson(currentAssignments, currentReplicaLogDirs));
System.out.printf("Proposed partition reassignment configuration%n%s%n",
formatAsReassignmentJson(proposedAssignments, Map.of()));
Review Comment:
open https://issues.apache.org/jira/browse/KAFKA-19637 as follow-up to make the assignment include directories
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19582 the current assignments shown by ReassignPartitionsCommand should include the log directories [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20319:
URL: https://github.com/apache/kafka/pull/20319#discussion_r2264592086
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -436,29 +437,50 @@ public void testGenerateAssignmentWithFewerBrokers() throws Exception {
@Test
public void testCurrentPartitionReplicaAssignmentToString() throws Exception {
- Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
-
- proposedParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
- proposedParts.put(new TopicPartition("bar", 0), List.of(7, 8, 9));
-
- Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
-
- currentParts.put(new TopicPartition("foo", 0), List.of(1, 2, 3));
- currentParts.put(new TopicPartition("foo", 1), List.of(4, 5, 6));
- currentParts.put(new TopicPartition("bar", 0), List.of(7, 8));
- currentParts.put(new TopicPartition("baz", 0), List.of(10, 11, 12));
+ try (MockAdminClient adminClient = new MockAdminClient.Builder()
+ .numBrokers(6)
+ .brokerLogDirs(List.of(
+ List.of("/tmp/broker0/logs"),
+ List.of("/tmp/broker1/logs"),
+ List.of("/tmp/broker2/logs"),
+ List.of("/tmp/broker3/logs"),
+ List.of("/tmp/broker4/logs"),
+ List.of("/tmp/broker5/logs")
+ ))
+ .build()
+ ) {
+
+ List<Node> brokers = adminClient.brokers();
+ adminClient.addTopic(false, "foo", List.of(
+ new TopicPartitionInfo(1, brokers.get(1),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)),
+ List.of(brokers.get(1), brokers.get(2), brokers.get(3)))
+ ), Map.of());
+
+ adminClient.addTopic(false, "bar", List.of(
+ new TopicPartitionInfo(0, brokers.get(4),
+ List.of(brokers.get(4), brokers.get(5)),
+ List.of(brokers.get(4), brokers.get(5)))
+ ), Map.of());
+
+ Map<TopicPartition, List<Integer>> proposedParts = new HashMap<>();
+ proposedParts.put(new TopicPartition("foo", 1), List.of(0, 1, 2));
+ proposedParts.put(new TopicPartition("bar", 0), List.of(3, 4, 5));
+
+ Map<TopicPartition, List<Integer>> currentParts = new HashMap<>();
+ currentParts.put(new TopicPartition("foo", 1), List.of(1, 2, 3));
+ currentParts.put(new TopicPartition("bar", 0), List.of(4, 5));
- assertEquals(String.join(System.lineSeparator(),
- "Current partition replica assignment",
- "",
- "{\"version\":1,\"partitions\":" +
- "[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[7,8],\"log_dirs\":[\"any\",\"any\"]}," +
- "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[4,5,6],\"log_dirs\":[\"any\",\"any\",\"any\"]}]" +
- "}",
- "",
- "Save this to use as the --reassignment-json-file option during rollback"),
- currentPartitionReplicaAssignmentToString(proposedParts, currentParts)
- );
+ assertEquals(String.join(System.lineSeparator(),
+ "Current partition replica assignment",
+ "",
+ "{\"version\":1,\"partitions\":[{\"topic\":\"bar\",\"partition\":0,\"replicas\":[4,5],\"log_dirs\":[\"/tmp/broker4/logs\",\"/tmp/broker4/logs\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}]}",
Review Comment:
It's fine for me right now, it's MockAdminClient's issue. I'll track this. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org