lianetm commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3723422992
##########
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:
uhm interesting point about the case of partition changing, agree with the
gap, but that's an issue in the "full" strategy as well, right? (send without
partition + default partitioning + no key + concurrent send with same default
partitioning that makes the partition change. Then append just continues
retrying if the partition changed, unbounded)
Here and again a few lines below:
https://github.com/apache/kafka/blob/6425799134313fe3216555b34cdb81b04bf9f102/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L322-L324
We probably need to consider enforcing a deadline for that partition-change
case too (just as we did on the extension path here). A separate PR/jira
better? given that the issue exists in both strategies (and the fix would go in
the default path RecordAccumulator and the new Chunk one too I expect)
--
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]
