hachikuji commented on a change in pull request #9406:
URL: https://github.com/apache/kafka/pull/9406#discussion_r509456514
##########
File path:
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
##########
@@ -2667,4 +2760,43 @@ private void assertFutureFailure(Future<?> future,
Class<? extends Exception> ex
}
}
+ private void createMockClientWithMaxFlightOneMetadataPending() {
+ client = new MockClient(time, metadata) {
Review comment:
Wonder if we should consider adding max inflight behavior directly to
`MockClient`. Seems like a notable difference from `NetworkClient`.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
##########
@@ -444,10 +444,25 @@ 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) {
+ if (!awaitNodeReady(targetNode, coordinatorType)) {
+ log.trace("Target node {} not ready within request
timeout, will retry when node is ready.", targetNode);
+ maybeFindCoordinatorAndRetry(nextRequestHandler);
+ return true;
+ }
+ } 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, will poll and
retry when until a node is ready.");
+ transactionManager.retry(nextRequestHandler);
+ client.poll(retryBackoffMs, time.milliseconds());
+ return true;
Review comment:
I agree polling seems sufficient. We will still have an opportunity to
refresh metadata if the current connection fails for some reason.
----------------------------------------------------------------
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
