lianetm commented on code in PR #23008:
URL: https://github.com/apache/kafka/pull/23008#discussion_r3779019724
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1556,6 +1560,49 @@ public void
testDrainWithANodeThatDoesntHostAnyPartitions() {
assertTrue(batches.get(node2.id()).isEmpty());
}
+ @Test
+ public void testRetryPolicy() {
+ RecordAccumulator accum = createTestRecordAccumulator(1024, 10 * 1024,
Compression.NONE, 0);
+ try {
+ long spent = time.milliseconds(); // a deadline that is
up: reached, so no time is left
+ long open = time.milliseconds() + 1000; // and one that is not
+
+ // The first pass may always run, whether or not there is time
left.
+ accum.throwIfNoMoreRetriesAllowed(/* firstPass */ true, spent,
false, topic);
+ accum.throwIfNoMoreRetriesAllowed(/* firstPass */ true, open,
false, topic);
+
+ // So may a retry, while there is time left.
+ accum.throwIfNoMoreRetriesAllowed(/* firstPass */ false, open,
false, topic);
+
+ // A retry that finds the deadline gone gives up, reporting the
cause the pass before it hit.
+ TimeoutException timeout = assertThrows(TimeoutException.class,
+ () -> accum.throwIfNoMoreRetriesAllowed(false, spent,
false, topic));
+ // BufferExhaustedException extends TimeoutException, so
assertThrows above would accept it too.
+ assertEquals(TimeoutException.class, timeout.getClass(),
timeout.getMessage());
Review Comment:
sure, done (on several msgs)
--
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]
