lianetm commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3761853619
##########
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:
Agreed, added a single mechanism to track the deadline across retries (all
retries).
The fix applies to ChunkedAccum only, but I defined the core funcs in the
base RecordAccum so we can reuse to bound retries on the full strategy too
(even though there it should be the partition change only, will file jira to
review that separately as it would touch the default path).
--
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]
