You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2026/06/10 10:33:58 UTC
[PR] KAFKA-20673: Skip stale-leader lookup retry while the AdminClient is closing (follow-up) [kafka]
lucasbru opened a new pull request, #22529:
URL: https://github.com/apache/kafka/pull/22529
Follow-up to KAFKA-20673 (#22493). That change lets the AdminClient recover a partition-leader request whose cached leader has left the cluster by sending the keys back to the lookup stage from the admin client thread. The re-lookup is enqueued through the runnable, which rejects new calls once close() has been initiated, failing them with "Cannot accept new calls when AdminClient is closing". As a result, a request that took this recovery path during the close(timeout) grace period would fail immediately instead of being retried until the deadline, even though the user asked for a grace period.
This change skips the recovery while the client is shutting down. handleNodeUnavailable returns false once a hard shutdown time has been set, so the call is left in pendingCalls and follows the normal close(timeout) handling: it can still be retried, or assigned if the broker reappears, within the shutdown grace period rather than being killed instantly. The behaviour outside of shutdown is unchanged.
--
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-20673: Skip stale-leader lookup retry while the AdminClient is closing (follow-up) [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22529:
URL: https://github.com/apache/kafka/pull/22529#discussion_r3387731409
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -5166,6 +5166,15 @@ void maybeRetry(long currentTimeMs, Throwable throwable) {
@Override
boolean handleNodeUnavailable(long currentTimeMs) {
+ // Don't intervene while the client is shutting down. Re-running the lookup enqueues
+ // new calls via runnable.call(), which are rejected during close ("Cannot accept new
+ // calls when AdminClient is closing"). Leaving the call in pendingCalls preserves the
+ // normal close(timeout) handling, so it can still be retried (or assigned, should the
+ // broker reappear) within the shutdown grace period instead of failing immediately.
+ if (hardShutdownTimeMs.get() != INVALID_SHUTDOWN_TIME) {
+ return false;
+ }
Review Comment:
The new shutdown guard (`hardShutdownTimeMs != INVALID_SHUTDOWN_TIME`) changes retry behavior specifically during `close(timeout)` (skipping the stale-leader lookup recovery). There doesn't appear to be a unit/integration test that exercises the stale-cached-leader recovery path while a `close(Duration)` is in progress, so this behavior could regress without detection. Consider adding a test that seeds the partition-leader cache with a broker that is removed from metadata, initiates a request, calls `close(Duration)` with a non-zero grace period, and asserts the call is not failed immediately with `IllegalStateException("Cannot accept new calls when AdminClient is closing.")` (it should instead follow the normal close/timeout path).
--
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-20673: Skip stale-leader lookup retry while the AdminClient is closing (follow-up) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #22529:
URL: https://github.com/apache/kafka/pull/22529#issuecomment-4716269079
@chia7712 Could you have look?
--
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-20673: Skip stale-leader lookup retry while the AdminClient is closing (follow-up) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #22529:
URL: https://github.com/apache/kafka/pull/22529
--
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-20673: Skip stale-leader lookup retry while the AdminClient is closing (follow-up) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22529:
URL: https://github.com/apache/kafka/pull/22529#issuecomment-4976206242
@lucasbru thanks for this patch. This approach is LGTM. Would you mind adding a small integration test? For example:
```java
public void testAdminClientCloseGracePeriodWithStaleLeader(ClusterInstance cluster) throws Exception {
try (Admin admin = cluster.admin()) {
String topic = "test-topic";
admin.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).all().get();
TopicDescription topicDescription = admin.describeTopics(List.of(topic))
.allTopicNames().get().get(topic);
int leaderId = topicDescription.partitions().get(0).leader().id();
var tp = new TopicPartition(topic, 0);
// update cache
admin.deleteRecords(Map.of(tp, RecordsToDelete.beforeOffset(0L))).all().get();
// cache miss
cluster.shutdownBroker(leaderId);
DeleteRecordsResult result = admin.deleteRecords(Map.of(tp, RecordsToDelete.beforeOffset(0L)));
admin.close(Duration.ofSeconds(5));
var e = Assertions.assertThrows(ExecutionException.class, () -> result.all().get());
Assertions.assertInstanceOf(TimeoutException.class, e.getCause());
}
}
```
--
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