You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Hongten (via GitHub)" <gi...@apache.org> on 2026/03/06 03:33:30 UTC
[PR] KAFKA-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Hongten opened a new pull request, #21654:
URL: https://github.com/apache/kafka/pull/21654
**Problem**
When a broker is down and executeAssignment() runs, incrementalAlterConfigs can throw TimeoutException because it tries to set broker-level throttles (inter-broker and log-dir) on brokers that are no longer live. The AdminClient sends config updates directly to each broker; if a broker is down, the request times out and the reassignment fails.
**Solution**
Apply throttle configuration only to brokers that are currently live. Live brokers are determined via adminClient.describeCluster().nodes(). Throttle configs are no longer sent to brokers that are not in this set, so reassignment can proceed even when some brokers are down.
I provided another solution before, Refer to https://github.com/apache/kafka/pull/21388, it will add a new parameter `--broker-list-without-throttle` to remove the broker, which is down.
The new solution is better I think.
**Changes**
1. getLiveBrokerIds() helper
Returns the set of broker IDs from describeCluster().nodes().
2. modifyInterBrokerThrottle()
Filters reassigningBrokers to only live brokers before calling incrementalAlterConfigs.
Returns early if no brokers are live.
3. modifyLogDirThrottle()
Filters movingBrokers to only live brokers before calling incrementalAlterConfigs.
Returns early if no brokers are live.
4. clearAllThrottles()
Clears broker-level throttles only for live brokers.
Uses retainAll(liveBrokers) so we never call incrementalAlterConfigs on down brokers.
**Testing**
Unit tests in ReassignPartitionsUnitTest:
1. testModifyInterBrokerThrottleSkipsDownBrokers – MockAdminClient with 3 brokers; modifyInterBrokerThrottle is called with brokers 0–3. Verifies throttle is applied only to 0, 1, 2 and broker 3 is skipped.
2. testModifyLogDirThrottleSkipsDownBrokers – Same setup for modifyLogDirThrottle.
3. testExecuteAssignmentWithThrottleWhenBrokerDown – Full executeAssignment with throttle on a 3-broker cluster; verifies reassignment succeeds and throttle is applied only to live brokers.
4. testModifyInterBrokerThrottleWhenAllBrokersDown – All reassigning brokers are down; verifies no exception and early return without calling incrementalAlterConfigs.
MockAdminClient with 3 brokers is used to simulate broker 3 being down (not in describeCluster().nodes()). Without the fix, incrementalAlterConfigs for broker 3 would fail with InvalidRequestException in the mock (or TimeoutException in a real cluster).
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903555855
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersToUpdate = reassigningBrokers.stream()
+ .filter(liveBrokers::contains)
Review Comment:
I will adjust.
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersToUpdate = reassigningBrokers.stream()
+ .filter(liveBrokers::contains)
+ .collect(Collectors.toSet());
+ if (brokersToUpdate.isEmpty()) {
+ return;
+ }
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
- reassigningBrokers.forEach(brokerId -> {
+ brokersToUpdate.forEach(brokerId -> {
Review Comment:
I will adjust.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903547854
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokers = new HashSet<>(liveBrokers);
targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ brokers.retainAll(liveBrokers);
Review Comment:
Yeah, I will adjust as below:
Simplified to use `liveBrokers` directly for clearing (the original addAll+retainAll was indeed equivalent to just liveBrokers).
Added logic to compute `downBrokers` = brokers from targetParts that are not in liveBrokers.
Added warning message: "Warning: Could not clear broker-level throttles on down broker(s) X. Throttles will be cleared when the broker comes back up."
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898467815
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersToUpdate = reassigningBrokers.stream()
+ .filter(liveBrokers::contains)
+ .collect(Collectors.toSet());
+ if (brokersToUpdate.isEmpty()) {
+ return;
+ }
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
- reassigningBrokers.forEach(brokerId -> {
+ brokersToUpdate.forEach(brokerId -> {
Review Comment:
Similar to above comment, we should print if we're unable to set or clear throttles for down brokers.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4045096983
Thanks @akhileshchg for the approval.
gentle ping @chia7712 @FrankYang0529 @lucasbru @showuon @lianetm
The review comments have been addressed. If the checks look good, could you please help merge this PR into trunk? thank you.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4949975959
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4677347127
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4030162162
> The PR looks good. Think the PR description is stale after the recent changes. Please update it. Have a couple of nits on the recent version. LGTM to me overall.
Thanks for LGTM and the feedback! @akhileshchg
Could you help to review the update of the nits?
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4009332236
cc @mimaison @chia7712 @showuon @lianetm This PR is ready for review, please take a look when you get a moment. 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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898472500
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -696,6 +717,82 @@ public void testModifyLogDirThrottle() throws Exception {
}
}
+ /**
+ * Test that modifyInterBrokerThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyInterBrokerThrottle(adminClient, Set.of(0, 1, 2, 3), 1000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyLogDirThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyLogDirThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyLogDirThrottle(adminClient, Set.of(0, 1, 2, 3), 2000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), -1, 2000);
+ }
+ }
+
+ /**
+ * Test that executeAssignment with throttle succeeds when some brokers are down.
+ * Uses MockAdminClient with 3 brokers; reassignment only involves live brokers (0, 1, 2).
+ * The throttle is applied only to live brokers via modifyInterBrokerThrottle/modifyLogDirThrottle.
+ */
+ @Test
+ public void testExecuteAssignmentWithThrottleWhenBrokerDown() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ addTopicsForThreeBrokers(adminClient);
+ String assignment = "{\"version\":1,\"partitions\":" +
+ "[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[0,1,2],\"log_dirs\":[\"any\",\"any\",\"any\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,0],\"log_dirs\":[\"any\",\"any\",\"any\"]}" +
+ "]}";
+ executeAssignment(adminClient, false, assignment, 1000L, 2000L, 10000L, Time.SYSTEM, false);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ }
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ // Only inter-broker throttle is applied; log-dir throttle is not set since log_dirs are all "any"
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyInterBrokerThrottle succeeds when all reassigning brokers are down.
+ * Should return early without calling incrementalAlterConfigs.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleWhenAllBrokersDown() throws Exception {
Review Comment:
Can we actually verify if incrementAlterConfigs was not called in 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2906855250
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
- List<Entry<TopicPartition, List<Integer>>> targetParts
+ static void clearAllThrottles(Admin adminClient,
+ List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
- targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersFromAssignment = new HashSet<>();
+ if (targetParts != null) {
+ for (Entry<TopicPartition, List<Integer>> t : targetParts) {
+ List<Integer> replicas = t.getValue();
+ if (replicas != null) {
+ brokersFromAssignment.addAll(replicas);
+ }
+ }
+ }
+ Set<Integer> downBrokers = brokersFromAssignment.stream()
+ .filter(b -> !liveBrokers.contains(b))
+ .collect(Collectors.toSet());
- System.out.printf("Clearing broker-level throttles on broker%s %s%n",
- brokers.size() == 1 ? "" : "s", brokers.stream().map(Object::toString).collect(Collectors.joining(",")));
- clearBrokerLevelThrottles(adminClient, brokers);
+ if (!downBrokers.isEmpty()) {
+ System.out.printf("Warning: Could not clear broker-level throttles on down broker%s %s. " +
+ "Throttles will be cleared when the broker comes back up.%n",
+ downBrokers.size() == 1 ? "" : "s",
Review Comment:
We should change the second sentence to `"Re-run --verify after the brokers are back up to
clear its throttles."`
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on PR #21654:
URL: https://github.com/apache/kafka/pull/21654#issuecomment-4025459475
The PR looks good. Think the PR description is stale after the recent changes. Please update it. Have a couple of nits on the recent version. LGTM to me overall.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898358508
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokers = new HashSet<>(liveBrokers);
targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ brokers.retainAll(liveBrokers);
Review Comment:
the `retainAll(liveBrokers)` after `addAll(t.getValue())` makes the addAll a no-op. If that is the intended behavior, we might as well simply use `liveBrokers`.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2909163059
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
Review Comment:
I think this solution should be work.
Restore `clearAllThrottles` as private and test via `verifyAssignment` instead of direct call. It will call `clearAllThrottles` when done.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898344529
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
Review Comment:
`executeAssignment` won't call clearAllThrottles. Only `--verify` does. If the intent is also to remove timeouts in `--verify`, please update the PR description.
Also, I think the PR is specifically to handle the cases when the few sources brokers are down and not actually the target brokers (because we do run verifyBrokerIds). This is also not clear in the PR description.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903514776
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -696,6 +717,82 @@ public void testModifyLogDirThrottle() throws Exception {
}
}
+ /**
+ * Test that modifyInterBrokerThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyInterBrokerThrottle(adminClient, Set.of(0, 1, 2, 3), 1000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyLogDirThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyLogDirThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyLogDirThrottle(adminClient, Set.of(0, 1, 2, 3), 2000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), -1, 2000);
+ }
+ }
+
+ /**
+ * Test that executeAssignment with throttle succeeds when some brokers are down.
+ * Uses MockAdminClient with 3 brokers; reassignment only involves live brokers (0, 1, 2).
+ * The throttle is applied only to live brokers via modifyInterBrokerThrottle/modifyLogDirThrottle.
+ */
+ @Test
+ public void testExecuteAssignmentWithThrottleWhenBrokerDown() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ addTopicsForThreeBrokers(adminClient);
+ String assignment = "{\"version\":1,\"partitions\":" +
+ "[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[0,1,2],\"log_dirs\":[\"any\",\"any\",\"any\"]}," +
+ "{\"topic\":\"foo\",\"partition\":1,\"replicas\":[1,2,0],\"log_dirs\":[\"any\",\"any\",\"any\"]}" +
+ "]}";
+ executeAssignment(adminClient, false, assignment, 1000L, 2000L, 10000L, Time.SYSTEM, false);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ }
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ // Only inter-broker throttle is applied; log-dir throttle is not set since log_dirs are all "any"
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyInterBrokerThrottle succeeds when all reassigning brokers are down.
+ * Should return early without calling incrementalAlterConfigs.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleWhenAllBrokersDown() throws Exception {
Review Comment:
Yeah, I will add such test.
My logic is like:
- Using a JDK dynamic proxy: wrap MockAdminClient with an Admin proxy that throws AssertionError if incrementalAlterConfigs is invoked.
- When modifyInterBrokerThrottle(adminThatFailsIfConfigAltered, Set.of(3), 1000) runs with all brokers down, it returns early without calling incrementalAlterConfigs.
- If our code had a bug and called incrementalAlterConfigs, the proxy would throw and the test would fail.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903552716
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
Review Comment:
`getLiveBrokerIds` will be called once in `executeAssignment` before the throttle block, and the result is passed to `modifyReassignmentThrottle` and `modifyLogDirThrottle`.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898475071
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -696,6 +717,82 @@ public void testModifyLogDirThrottle() throws Exception {
}
}
+ /**
+ * Test that modifyInterBrokerThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyInterBrokerThrottle(adminClient, Set.of(0, 1, 2, 3), 1000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyLogDirThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyLogDirThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyLogDirThrottle(adminClient, Set.of(0, 1, 2, 3), 2000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), -1, 2000);
+ }
+ }
+
+ /**
+ * Test that executeAssignment with throttle succeeds when some brokers are down.
+ * Uses MockAdminClient with 3 brokers; reassignment only involves live brokers (0, 1, 2).
+ * The throttle is applied only to live brokers via modifyInterBrokerThrottle/modifyLogDirThrottle.
+ */
+ @Test
+ public void testExecuteAssignmentWithThrottleWhenBrokerDown() throws Exception {
Review Comment:
The test name says the broker is down, but it seems all the 3 brokers are live in 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903521328
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
@@ -696,6 +717,82 @@ public void testModifyLogDirThrottle() throws Exception {
}
}
+ /**
+ * Test that modifyInterBrokerThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyInterBrokerThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyInterBrokerThrottle(adminClient, Set.of(0, 1, 2, 3), 1000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), 1000, -1);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), 1000, -1);
+ }
+ }
+
+ /**
+ * Test that modifyLogDirThrottle only applies throttle to live brokers.
+ * When broker 3 is down (not in describeCluster), it should be skipped to avoid TimeoutException.
+ */
+ @Test
+ public void testModifyLogDirThrottleSkipsDownBrokers() throws Exception {
+ try (MockAdminClient adminClient = new MockAdminClient.Builder().numBrokers(3).build()) {
+ modifyLogDirThrottle(adminClient, Set.of(0, 1, 2, 3), 2000);
+
+ List<ConfigResource> brokers = new ArrayList<>();
+ for (int i = 0; i < 3; i++)
+ brokers.add(new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(i)));
+ Map<ConfigResource, Config> results = adminClient.describeConfigs(brokers).all().get();
+ verifyBrokerThrottleResults(results.get(brokers.get(0)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(1)), -1, 2000);
+ verifyBrokerThrottleResults(results.get(brokers.get(2)), -1, 2000);
+ }
+ }
+
+ /**
+ * Test that executeAssignment with throttle succeeds when some brokers are down.
+ * Uses MockAdminClient with 3 brokers; reassignment only involves live brokers (0, 1, 2).
+ * The throttle is applied only to live brokers via modifyInterBrokerThrottle/modifyLogDirThrottle.
+ */
+ @Test
+ public void testExecuteAssignmentWithThrottleWhenBrokerDown() throws Exception {
Review Comment:
I will make it more clear.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903535225
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
Review Comment:
Updated the PR description to clarify that clearAllThrottles is called by `--verify` when reassignment completes, and the fix also prevents TimeoutException during `--verify`.
Clarified that the fix handles down source brokers (replicas we're moving FROM). Target brokers must pass `verifyBrokerIds` and thus exist in the cluster. The typical scenario: source broker goes down after reassignment starts; we skip setting/clearing throttle on it so the operation can proceed.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2903526545
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
Review Comment:
I will add testClearAllThrottlesSkipsDownBrokers().
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2906877339
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
Review Comment:
Nit: Can we use outer methods if it is not too difficult instead of changing the visibility of the method?
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2906782770
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
- List<Entry<TopicPartition, List<Integer>>> targetParts
+ static void clearAllThrottles(Admin adminClient,
+ List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
- targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersFromAssignment = new HashSet<>();
+ if (targetParts != null) {
+ for (Entry<TopicPartition, List<Integer>> t : targetParts) {
+ List<Integer> replicas = t.getValue();
+ if (replicas != null) {
+ brokersFromAssignment.addAll(replicas);
+ }
+ }
+ }
+ Set<Integer> downBrokers = brokersFromAssignment.stream()
+ .filter(b -> !liveBrokers.contains(b))
+ .collect(Collectors.toSet());
- System.out.printf("Clearing broker-level throttles on broker%s %s%n",
- brokers.size() == 1 ? "" : "s", brokers.stream().map(Object::toString).collect(Collectors.joining(",")));
- clearBrokerLevelThrottles(adminClient, brokers);
+ if (!downBrokers.isEmpty()) {
+ System.out.printf("Warning: Could not clear broker-level throttles on down broker%s %s. " +
+ "Throttles will be cleared when the broker comes back up.%n",
Review Comment:
Nit: Throttles can be cleared using `--verify` after the broker comes back up.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898462039
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
Review Comment:
For one execute command, we're calling describe RPC twice for InterBrokerThrottle and LogDirThrottle. We should call this once in the parent executeAssignment method and pass it down.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898471205
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java:
##########
Review Comment:
There is no test for clearAllThrottles.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898362726
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokers = new HashSet<>(liveBrokers);
targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ brokers.retainAll(liveBrokers);
Review Comment:
It also makes sense to print if there's a down-broker that had the thresholds set when it was live, but --verify could not clear them.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898464056
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -1192,8 +1207,15 @@ static void modifyInterBrokerThrottle(Admin adminClient,
Set<Integer> reassigningBrokers,
long interBrokerThrottle) throws ExecutionException, InterruptedException {
if (interBrokerThrottle >= 0) {
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersToUpdate = reassigningBrokers.stream()
+ .filter(liveBrokers::contains)
Review Comment:
The live broker set is fetched at one point in time, but the incrementalAlterConfigs call happens later. A broker could go down between these two calls, or come back up. This is inherent to the approach and probably acceptable, but worth acknowledging.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2898344529
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -499,8 +510,10 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
private static void clearAllThrottles(Admin adminClient,
Review Comment:
`executeAssignment` won't call clearAllThrottles. Only `--verify` does. If the intent is also to remove timeouts in `--verify`, please update the PR description.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2909163059
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
Review Comment:
I think this solution should be work.
Restore `clearAllThrottles` as private and test via `verifyAssignment` instead of direct call.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "akhileshchg (via GitHub)" <gi...@apache.org>.
akhileshchg commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2906782770
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
- List<Entry<TopicPartition, List<Integer>>> targetParts
+ static void clearAllThrottles(Admin adminClient,
+ List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
- targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersFromAssignment = new HashSet<>();
+ if (targetParts != null) {
+ for (Entry<TopicPartition, List<Integer>> t : targetParts) {
+ List<Integer> replicas = t.getValue();
+ if (replicas != null) {
+ brokersFromAssignment.addAll(replicas);
+ }
+ }
+ }
+ Set<Integer> downBrokers = brokersFromAssignment.stream()
+ .filter(b -> !liveBrokers.contains(b))
+ .collect(Collectors.toSet());
- System.out.printf("Clearing broker-level throttles on broker%s %s%n",
- brokers.size() == 1 ? "" : "s", brokers.stream().map(Object::toString).collect(Collectors.joining(",")));
- clearBrokerLevelThrottles(adminClient, brokers);
+ if (!downBrokers.isEmpty()) {
+ System.out.printf("Warning: Could not clear broker-level throttles on down broker%s %s. " +
+ "Throttles will be cleared when the broker comes back up.%n",
Review Comment:
Nit: Throttles can be cleared using `--verify` after the broker comes back up.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "Hongten (via GitHub)" <gi...@apache.org>.
Hongten commented on code in PR #21654:
URL: https://github.com/apache/kafka/pull/21654#discussion_r2909125828
##########
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java:
##########
@@ -490,26 +491,63 @@ static String replicaMoveStatesToString(Map<TopicPartitionReplica, LogDirMoveSta
return bld.stream().collect(Collectors.joining(System.lineSeparator()));
}
+ /**
+ * Get the set of broker IDs that are currently live in the cluster.
+ * Only live brokers can be successfully configured via incrementalAlterConfigs.
+ *
+ * @param adminClient The AdminClient to use.
+ * @return The set of live broker IDs.
+ */
+ private static Set<Integer> getLiveBrokerIds(Admin adminClient) throws ExecutionException, InterruptedException {
+ return adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
+ }
+
/**
* Clear all topic-level and broker-level throttles.
+ * Only clears throttles on live brokers to avoid TimeoutException when some brokers are down.
+ * Called by --verify when reassignment is complete.
*
* @param adminClient The AdminClient to use.
* @param targetParts The target partitions loaded from the JSON file.
*/
- private static void clearAllThrottles(Admin adminClient,
- List<Entry<TopicPartition, List<Integer>>> targetParts
+ static void clearAllThrottles(Admin adminClient,
+ List<Entry<TopicPartition, List<Integer>>> targetParts
) throws ExecutionException, InterruptedException {
- Set<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toSet());
- targetParts.forEach(t -> brokers.addAll(t.getValue()));
+ Set<Integer> liveBrokers = getLiveBrokerIds(adminClient);
+ Set<Integer> brokersFromAssignment = new HashSet<>();
+ if (targetParts != null) {
+ for (Entry<TopicPartition, List<Integer>> t : targetParts) {
+ List<Integer> replicas = t.getValue();
+ if (replicas != null) {
+ brokersFromAssignment.addAll(replicas);
+ }
+ }
+ }
+ Set<Integer> downBrokers = brokersFromAssignment.stream()
+ .filter(b -> !liveBrokers.contains(b))
+ .collect(Collectors.toSet());
- System.out.printf("Clearing broker-level throttles on broker%s %s%n",
- brokers.size() == 1 ? "" : "s", brokers.stream().map(Object::toString).collect(Collectors.joining(",")));
- clearBrokerLevelThrottles(adminClient, brokers);
+ if (!downBrokers.isEmpty()) {
+ System.out.printf("Warning: Could not clear broker-level throttles on down broker%s %s. " +
+ "Throttles will be cleared when the broker comes back up.%n",
+ downBrokers.size() == 1 ? "" : "s",
Review Comment:
Let me change it.
--
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-13392 Fix incrementalAlterConfigs timeout when broker is down [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #21654: KAFKA-13392 Fix incrementalAlterConfigs timeout when broker is down
URL: https://github.com/apache/kafka/pull/21654
--
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