dajac commented on a change in pull request #9406:
URL: https://github.com/apache/kafka/pull/9406#discussion_r508626975
##########
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:
@rajinisivaram It seems that a non-null `nextRequestHandler` would have
been removed from the queue:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L872.
I've got the impression that `InitProducerIdHandler` may hit that code path
as well as it could have a null `coordinatorType` when the transaction id is
not set (idempotent producer). I think that we could unit test this actually by
reusing your `testInitProducerIdWithMaxInFlightOne` but without setting the
transactional id. Should we add this test to be on the safe side?
----------------------------------------------------------------
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
