lucasbru commented on code in PR #22493:
URL: https://github.com/apache/kafka/pull/22493#discussion_r3374911670
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -5151,6 +5163,26 @@ void maybeRetry(long currentTimeMs, Throwable throwable)
{
super.maybeRetry(currentTimeMs, throwable);
}
}
+
+ @Override
+ boolean handleNodeUnavailable(long currentTimeMs) {
+ OptionalInt brokerId = spec.scope.destinationBrokerId();
+ if (brokerId.isPresent()
+ && metadataManager.isReady()
+ && metadataManager.nodeById(brokerId.getAsInt()) ==
null) {
+ // The fulfillment target broker is no longer present in
the cluster metadata.
+ // This happens when a stale entry in the partition leader
cache points at a
+ // broker that has since left the cluster. Send the keys
back to the lookup
+ // stage so the leader can be re-resolved, rather than
waiting for the request
+ // deadline to expire without ever issuing another lookup.
+ log.debug("Broker {} for {} is no longer in the cluster
metadata; retrying lookup.",
+ brokerId.getAsInt(), spec.name);
+ driver.retryLookup(currentTimeMs, spec);
+ maybeSendRequests(driver, currentTimeMs);
+ return true;
+ }
+ return false;
Review Comment:
Good catch. Fixed: handleNodeUnavailable now only retries the lookup when
retryLookup reports it could move the keys back. For StaticBrokerStrategy
(describeProducers with an explicit broker id) it's a no-op, so the call is
left pending until the deadline as before, rather than looping and exhausting
the retry budget. Added testRetryLookupIsNoOpForFixedBrokerStrategy to cover 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
