You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/04/05 01:25:00 UTC
[GitHub] [kafka] hachikuji commented on a diff in pull request #11991: KAFKA-13794: Fix comparator of inflightBatchesBySequence in TransactionManager
hachikuji commented on code in PR #11991:
URL: https://github.com/apache/kafka/pull/11991#discussion_r842248811
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -184,16 +184,22 @@ private void startSequencesAtBeginning(TopicPartition topicPartition, ProducerId
// responses which are due to the retention period elapsing, and those which are due to actual lost data.
private long lastAckedOffset;
+ private final Comparator<ProducerBatch> producerBatchComparator = (b1, b2) -> {
Review Comment:
nit: could this be static? It doesn't have any state.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -674,6 +674,61 @@ public void testBatchCompletedAfterProducerReset() {
assertNull(transactionManager.nextBatchBySequence(tp0));
}
+ @Test
+ public void testDuplicateSequenceAfterProducerReset() throws Exception {
+ initializeTransactionManager(Optional.empty());
+ initializeIdempotentProducerId(producerId, epoch);
+
+ Metrics metrics = new Metrics(time);
+
+ RecordAccumulator accumulator = new RecordAccumulator(logContext, 16 * 1024, CompressionType.NONE, 0, 0L,
+ 15000, metrics, "", time, apiVersions, transactionManager,
+ new BufferPool(1024 * 1024, 16 * 1024, metrics, time, ""));
+
+ Sender sender = new Sender(logContext, this.client, this.metadata, accumulator, false,
+ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), this.time, 10000,
+ 0, transactionManager, apiVersions);
+
+ assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());
+
+ Future<RecordMetadata> responseFuture1 = accumulator.append(tp0, time.milliseconds(), "1".getBytes(), "1".getBytes(), Record.EMPTY_HEADERS,
+ null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
+ sender.runOnce();
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(10000); // request time out
+ sender.runOnce();
+ assertEquals(0, client.inFlightRequestCount());
+ assertTrue(transactionManager.hasInflightBatches(tp0));
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+ sender.runOnce(); // retry
+ assertEquals(1, client.inFlightRequestCount());
+ assertTrue(transactionManager.hasInflightBatches(tp0));
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(5000); // delivery time out
+ sender.runOnce(); // expired in accumulator
+ assertFalse(transactionManager.hasInFlightRequest());
+ assertEquals(1, client.inFlightRequestCount()); // not reaching request timeout, so still in flight
+
+ sender.runOnce(); // bump the epoch
+ assertEquals(epoch + 1, transactionManager.producerIdAndEpoch().epoch);
+ assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());
+
+ Future<RecordMetadata> responseFuture2 = accumulator.append(tp0, time.milliseconds(), "2".getBytes(), "2".getBytes(), Record.EMPTY_HEADERS,
+ null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
+ sender.runOnce();
+ sender.runOnce();
+ assertEquals(0, transactionManager.firstInFlightSequence(tp0));
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(5000); // request time out again
+ sender.runOnce();
+ assertTrue(transactionManager.hasInflightBatches(tp0)); // the latter batch failed and retried
+ assertTrue(responseFuture1.isDone());
Review Comment:
nit: could we move this after line 710? It makes the test a little easier to understand if we see when the first send fails. Also, maybe we could use `TestUtils.assertFutureThrows(responseFuture1, TimeoutException.class)` to make the timeout expectation explicit?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -674,6 +674,61 @@ public void testBatchCompletedAfterProducerReset() {
assertNull(transactionManager.nextBatchBySequence(tp0));
}
+ @Test
+ public void testDuplicateSequenceAfterProducerReset() throws Exception {
+ initializeTransactionManager(Optional.empty());
+ initializeIdempotentProducerId(producerId, epoch);
+
+ Metrics metrics = new Metrics(time);
+
+ RecordAccumulator accumulator = new RecordAccumulator(logContext, 16 * 1024, CompressionType.NONE, 0, 0L,
+ 15000, metrics, "", time, apiVersions, transactionManager,
+ new BufferPool(1024 * 1024, 16 * 1024, metrics, time, ""));
+
+ Sender sender = new Sender(logContext, this.client, this.metadata, accumulator, false,
+ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), this.time, 10000,
+ 0, transactionManager, apiVersions);
+
+ assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());
+
+ Future<RecordMetadata> responseFuture1 = accumulator.append(tp0, time.milliseconds(), "1".getBytes(), "1".getBytes(), Record.EMPTY_HEADERS,
+ null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
+ sender.runOnce();
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(10000); // request time out
Review Comment:
nit: could introduce a variable `requestTimeoutMs` and drop the comment
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -674,6 +674,61 @@ public void testBatchCompletedAfterProducerReset() {
assertNull(transactionManager.nextBatchBySequence(tp0));
}
+ @Test
+ public void testDuplicateSequenceAfterProducerReset() throws Exception {
+ initializeTransactionManager(Optional.empty());
+ initializeIdempotentProducerId(producerId, epoch);
+
+ Metrics metrics = new Metrics(time);
+
+ RecordAccumulator accumulator = new RecordAccumulator(logContext, 16 * 1024, CompressionType.NONE, 0, 0L,
+ 15000, metrics, "", time, apiVersions, transactionManager,
+ new BufferPool(1024 * 1024, 16 * 1024, metrics, time, ""));
+
+ Sender sender = new Sender(logContext, this.client, this.metadata, accumulator, false,
+ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), this.time, 10000,
+ 0, transactionManager, apiVersions);
+
+ assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());
+
+ Future<RecordMetadata> responseFuture1 = accumulator.append(tp0, time.milliseconds(), "1".getBytes(), "1".getBytes(), Record.EMPTY_HEADERS,
+ null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
+ sender.runOnce();
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(10000); // request time out
+ sender.runOnce();
+ assertEquals(0, client.inFlightRequestCount());
+ assertTrue(transactionManager.hasInflightBatches(tp0));
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+ sender.runOnce(); // retry
+ assertEquals(1, client.inFlightRequestCount());
+ assertTrue(transactionManager.hasInflightBatches(tp0));
+ assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
+
+ time.sleep(5000); // delivery time out
+ sender.runOnce(); // expired in accumulator
+ assertFalse(transactionManager.hasInFlightRequest());
Review Comment:
The behavior here puzzled me a little when I was trying to understand the test. Would a comment like this help?
```
// The retry request will remain inflight until the request timeout
// is reached even though the delivery timeout has been reached and
// the future has completed exceptionally.
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org