rajinisivaram commented on a change in pull request #9406:
URL: https://github.com/apache/kafka/pull/9406#discussion_r508583528
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
##########
@@ -444,10 +444,20 @@ private boolean maybeSendAndPollTransactionalRequest() {
AbstractRequest.Builder<?> requestBuilder =
nextRequestHandler.requestBuilder();
Node targetNode = null;
try {
- targetNode = awaitNodeReady(nextRequestHandler.coordinatorType());
- if (targetNode == null) {
+ FindCoordinatorRequest.CoordinatorType coordinatorType =
nextRequestHandler.coordinatorType();
+ targetNode = coordinatorType != null ?
+ transactionManager.coordinator(coordinatorType) :
+ client.leastLoadedNode(time.milliseconds());
+ if (targetNode != null) {
+ awaitNodeReady(targetNode, coordinatorType);
+ } else if (coordinatorType != null) {
+ log.trace("Coordinator not known for {}, will retry {} after
finding coordinator.", coordinatorType, requestBuilder.apiKey());
maybeFindCoordinatorAndRetry(nextRequestHandler);
return true;
+ } else {
+ log.trace("No nodes available to send requests, polling until
a node is ready.");
+ client.poll(retryBackoffMs, time.milliseconds());
+ return true;
Review comment:
@dajac Thanks for the review. As far as I can tell, `nextRequestHandler`
returned by `transactionManager.nextRequest` would have retained the request in
this case since it does a peek and removes the element only under certain
conditions. I was trying to limit the amount of change to the minimal necessary
to handle max.in.flight=1, do you think we should do more to look into the
other cases?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
