dajac commented on a change in pull request #9406:
URL: https://github.com/apache/kafka/pull/9406#discussion_r508473537
##########
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:
Don't we loose the current `nextRequestHandler` when we end up in this
branch?
For instance, if `nextRequestHandler` is a `FindCoordinatorHandler` and
there a no nodes available, `targetNode` is `null` and `coordinatorType` is
`null` as well so we end up here and poll. We don't do anything with
`nextRequestHandler` and return so it is gone. I suppose that it is not an
issue for `FindCoordinatorHandler` as a new one will be enqueued automatically
when another `TxnRequestHandler` handler is processed and the coordinator is
unknown.
We may want to push back `nextRequestHandler` to the queue with
`transactionManager.retry(nextRequestHandler)` in oder to handle all the cases.
I am not sure if that could really happen with any other `TxnRequestHandler`
type though. What do you 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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
