lianetm commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3715577942
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:
##########
@@ -122,11 +123,17 @@ public RecordAppendResult append(String topic,
// with that size. Set and cleared together across retries; null when
none is held.
NewBatchBuffer newBatch = null;
List<ByteBuffer> extensionChunks = null;
- // Budget shared by every blocking acquisition this append makes, so
the total blocking time
- // stays within maxTimeToBlock. The full strategy holds its one buffer
across retries and so
- // blocks at most once; this loop can release the chunks it acquired
(when a concurrent
- // appender created a batch to extend instead) and block again on a
later iteration.
- long remainingTimeToBlock = maxTimeToBlock;
+
+ // Bounds how long this append waits for memory, so it stays within
maxTimeToBlock, even across retries.
+ // E.g., this loop may release the chunks it acquired (when a
concurrent appender created a batch to extend instead)
+ // 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 throwIfExtensionBudgetSpent), to avoid retrying it
continuously with no bound.
+ boolean extensionAcquireFailed = false;
Review Comment:
yeap, good catch. Changed to check the deadline on all extension retries
(failed or successful).
Added it after the `tryAppend` though (not at the top of the loop), just to
allow an append that can fit into the open batch without extension. Also added
a test for this `testSucceedingExtensionRetriesStopOnceMaxBlockTimeIsUsedUp`
--
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]
