junrao commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3723619113
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:
##########
@@ -129,10 +129,11 @@ public RecordAppendResult append(String topic,
// and block again on a later iteration, so the blocking acquire is
given whatever is left of the deadline.
long deadlineMs = maxTimeToBlock > Long.MAX_VALUE - nowMs ?
Long.MAX_VALUE : nowMs + maxTimeToBlock;
- // Set once the extension acquire has failed on an exhausted pool.
That acquire is non-blocking, so we
- // always allow a first attempt (even with max.block.ms 0), and only
check retries of it against the
- // deadline (see throwIfExtensionBudgetExceeded), to avoid retrying it
continuously with no bound.
- boolean extensionAcquireFailed = false;
+ // Set once this append took the extension path, so any later pass
through it is a retry: the acquire failed,
+ // or succeeded without getting the record appended (not enough
capacity by then, or no longer the batch it was
+ // sized against). Neither blocks, so a retry never times out on its
own and is checked against the deadline
+ // based on this flag. A first attempt always runs, even with
max.block.ms 0.
+ boolean extensionRetried = false;
Review Comment:
The partition changing case applies to both strategies. I am just wondering
if we can have a single mechanism to handle all cases of potential infinite
loop, instead of having special mechanisms for each individual case.
--
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]
