junrao commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3762560471
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1556,6 +1560,59 @@ public void
testDrainWithANodeThatDoesntHostAnyPartitions() {
assertTrue(batches.get(node2.id()).isEmpty());
}
+ /**
+ * The retry policy that bounds an append loop, on its own and without the
interleavings needed to reach
+ * each branch through append: the first pass and one pass past the
deadline are free, passes inside the
+ * deadline are always free, and giving up reports the cause of the pass
that gave up.
+ */
+ @Test
+ public void testRetryPolicy() {
+ RecordAccumulator accum = createTestRecordAccumulator(1024, 10 * 1024,
Compression.NONE, 0);
+ try {
+ RecordAccumulator.AppendAttemptState firstAttempt =
RecordAccumulator.AppendAttemptState.FIRST_ATTEMPT;
+ RecordAccumulator.AppendAttemptState retrying =
RecordAccumulator.AppendAttemptState.RETRYING;
+ RecordAccumulator.AppendAttemptState expired =
RecordAccumulator.AppendAttemptState.RETRIES_EXPIRED;
+ long spent = time.milliseconds(); // a deadline that
has already passed
+ long open = time.milliseconds() + 1000; // and one that has
not
+
+ // The first pass may always run, whether or not there is time
left.
+ assertEquals(retrying,
accum.throwIfNoMoreRetriesAllowed(firstAttempt, spent, false, topic));
+ assertEquals(retrying,
accum.throwIfNoMoreRetriesAllowed(firstAttempt, open, false, topic));
+
+ // Inside the deadline a retry is free and stays a retry, so the
free pass is still in hand.
+ assertEquals(retrying, accum.throwIfNoMoreRetriesAllowed(retrying,
open, false, topic));
+
+ // Past the deadline, exactly one more pass.
+ assertEquals(expired, accum.throwIfNoMoreRetriesAllowed(retrying,
spent, false, topic));
+
+ // And then it gives up, reporting the cause the last pass hit.
+ TimeoutException timeout = assertThrows(TimeoutException.class,
+ () -> accum.throwIfNoMoreRetriesAllowed(expired, spent,
false, topic));
+ assertEquals(TimeoutException.class, timeout.getClass(),
timeout.getMessage());
+ assertTrue(timeout.getMessage().contains("kept restarting"),
timeout.getMessage());
Review Comment:
Should we use `retry` consistently instead of having both `restart` and
`retry`?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -382,6 +401,70 @@ protected RecordAppendResult
updatePartitionInfoOnAppend(RecordAppendResult appe
return appendResult;
}
+ /**
+ * Adds {@code maxTimeToBlock} to {@code nowMs}, capped at {@link
Long#MAX_VALUE} so that a large
+ * {@code max.block.ms} cannot overflow into a deadline in the past.
+ */
+ protected static long appendDeadlineMs(long nowMs, long maxTimeToBlock) {
+ return maxTimeToBlock > Long.MAX_VALUE - nowMs ? Long.MAX_VALUE :
nowMs + maxTimeToBlock;
+ }
+
+ /**
+ * What is left of {@code deadlineMs} to wait for memory. An acquire given
this rather than the raw
+ * {@code max.block.ms} has the time already spent retrying counted
against its wait. Reads the clock, not a
+ * cached {@code nowMs}, which retries that acquired nothing never refresh.
+ * <p>
+ * Only {@link ChunkedRecordAccumulator#append} uses it so far; {@link
#append} still passes the raw value.
+ */
+ protected long remainingTimeToBlockMs(long deadlineMs) {
+ return Math.max(0L, deadlineMs - time.milliseconds());
+ }
+
+ /**
+ * Decide whether an append pass may run, and throw if it may not because
no time is left.
+ * <ul>
+ * <li>the first pass is always allowed — the deadline is not even read,
so an append that completes in one
+ * pass never depends on the clock;</li>
+ * <li>retries are allowed while there is time left before {@code
deadlineMs};</li>
+ * <li>after the deadline, one more retry is allowed, since it may need no
memory at all and
Review Comment:
Intuitively, this feels weird. Why do we want to allow an extra retry after
the deadline has passed? If max.block.ms > 0, this is not needed since it's
covered by deadline. If max.block.ms=0, does one more retry guarantee success
when the blocking part is never hit?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:
##########
@@ -122,14 +122,22 @@ 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;
+
+ // This append's share of max.block.ms, as an absolute deadline.
Blocking allocations bound themselves against it.
+ // Retries that never block are bounded by it through
throwIfNoMoreRetriesAllowed: they are allowed while
+ // time is left. All retries bounded consistently (retry failed
extension, retry on partition change).
+ long deadlineMs = appendDeadlineMs(nowMs, maxTimeToBlock);
+ AppendAttemptState attemptState = AppendAttemptState.FIRST_ATTEMPT;
+ // Whether the non-blocking extension was denied memory on the pass
that just ended (only
+ // memory exhaustion case a pass can survive because it's
non-blocking, all others throw).
+ // Cleared once the next pass has read it, so it can only ever
describe the pass immediately before.
+ boolean nonBlockingAllocationDeniedMemory = false;
Review Comment:
nonBlockingAllocationDeniedMemory => nonBlockingMemoryAllocationDenied ?
--
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]
