junrao commented on code in PR #23234:
URL: https://github.com/apache/kafka/pull/23234#discussion_r3847836682
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/ProducerStateManagerTest.java:
##########
@@ -1199,6 +1206,107 @@ public void
testAllowNonZeroSequenceForTransactionsV1WithEmptyState() {
);
}
+ @Test
+ public void testRejectNonZeroFirstSequenceWithEmptyStateOnEmptyLog() {
+ // KAFKA-15591: A producer with no state must start at sequence 0 on a
partition which has never
+ // contained any records, since no state can have been lost. A
non-zero first sequence means the
+ // request arrived out of order, and accepting it would permanently
prevent the earlier request
+ // from being appended
+ ProducerAppendInfo laterRequest = new ProducerAppendInfo(
+ partition,
+ producerId,
+ ProducerStateEntry.empty(producerId),
+ AppendOrigin.CLIENT,
+ null,
+ true
+ );
+
+ OutOfOrderSequenceException exception =
assertThrows(OutOfOrderSequenceException.class,
+ () -> laterRequest.appendDataBatch(
+ epoch,
+ 17,
+ 21,
+ time.milliseconds(),
+ new LogOffsetMetadata(0L), 4L, false)
+ );
+ assertTrue(exception.getMessage().contains("Expected sequence 0 for a
producer with no state on a partition with no records"));
+
+ // The retried earlier request starting at sequence 0 is accepted
+ ProducerAppendInfo earlierRequest = new ProducerAppendInfo(
Review Comment:
laterRequest and earlierRequest are weird names here. They don't contain
anything specific to the request. Request specific params are passed into
appendDataBatch explicitly. They probably should be named sth like
currentProducerAppendInfo and just be created once.
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -1259,6 +1259,63 @@ public void testNonSequentialAppend() throws IOException
{
assertThrows(OutOfOrderSequenceException.class, () ->
log.appendAsLeader(nextRecords, 0));
}
+ @Test
+ public void testRejectOutOfOrderFirstRequestOnNewlyCreatedLog() throws
IOException {
+ // KAFKA-15591: A producer with multiple in-flight produce requests on
a newly created partition sends
+ // request A (sequences 0-3) and request B (sequences 4-5). Because
topic creation occurs asynchronously,
+ // request A can fail with NOT_LEADER_OR_FOLLOWER briefly because the
broker has not yet completed
+ // the topic creation, so request B is the first to reach the log. If
B were accepted, every retry of A
+ // would fail with OUT_OF_ORDER_SEQUENCE_NUMBER until it expires,
losing its records.
+ UnifiedLog log = createLog(logDir, new LogConfig(new Properties()));
+ long pid = 1L;
+ short epoch = 0;
+
+ MemoryRecords requestB = LogTestUtils.records(
+ List.of(new SimpleRecord("a".getBytes(), "b".getBytes()),
+ new SimpleRecord("a".getBytes(), "b".getBytes())),
+ pid, epoch, 4, 0L);
+ assertThrows(OutOfOrderSequenceException.class, () ->
log.appendAsLeader(requestB, 0));
+
+ MemoryRecords requestA = LogTestUtils.records(
+ List.of(new SimpleRecord("a".getBytes(), "b".getBytes()),
+ new SimpleRecord("a".getBytes(), "b".getBytes()),
+ new SimpleRecord("a".getBytes(), "b".getBytes()),
+ new SimpleRecord("a".getBytes(), "b".getBytes())),
+ pid, epoch, 0, 0L);
+ log.appendAsLeader(requestA, 0);
+
+ log.appendAsLeader(requestB, 0);
+ assertEquals(6L, log.logEndOffset());
+ }
+
+ @Test
+ public void testNonZeroFirstSequenceAcceptedAfterProducerStateExpiration()
throws IOException {
+ // KAFKA-15591: Once records exist in the log, a producer with no
state may start at a non-zero sequence.
+ // Its state may have legitimately been lost, such as through producer
expiration.
+ ProducerStateManagerConfig customPSMConfig = new
ProducerStateManagerConfig(200, false);
+ int producerIdExpirationCheckIntervalMs = 100;
+
+ LogConfig logConfig = new
LogTestUtils.LogConfigBuilder().segmentBytes(TEN_KB).build();
+ UnifiedLog log = createLog(logDir, logConfig, 0L, 0L, brokerTopicStats,
+ mockTime.scheduler, mockTime, customPSMConfig, true,
Optional.empty(), false,
+ producerIdExpirationCheckIntervalMs);
+ long pid = 1L;
+ short epoch = 0;
+
+ log.appendAsLeader(LogTestUtils.records(List.of(new
SimpleRecord("foo".getBytes())),
+ pid, epoch, 0, 0L), 0);
+ assertEquals(Set.of(pid),
log.activeProducersWithLastSequence().keySet());
+
+ mockTime.sleep(producerIdExpirationCheckIntervalMs);
+ mockTime.sleep(producerIdExpirationCheckIntervalMs);
Review Comment:
Could we just sleep once with producerIdExpirationMs?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -1532,6 +1532,95 @@ public void
testExpiryOfFirstBatchShouldCauseEpochBumpIfFutureBatchesFail() thro
assertFalse(transactionManager.hasUnresolvedSequence(tp0));
}
+ @Test
+ public void
testEpochBumpWhenOutOfOrderBatchesRetriedAndFirstBatchExpires() throws
Exception {
+ // KAFKA-15591: Tests the situation where a producer sends a sequence
of requests to a newly created
+ // partition before the creation of the partition has entirely
completed. The first request arrives
+ // before the partition is ready and fails with NOT_LEADER_OR_FOLLOWER
without reaching the log.
+ // The next two requests arrive after the partition creation is
complete, but the sequence numbers
+ // do not start at zero so the requests fail with
OUT_OF_ORDER_SEQUENCE_NUMBER. The first request
+ // could still fill the sequence gap, but it expires before being
retried. Once the first request
+ // expires, the producer bumps the epoch, renumbers the remaining
requests from sequence 0 and
+ // sends them again.
+ final long producerId = 343434L;
+ TransactionManager transactionManager = createTransactionManager();
+ setupWithTransactionState(transactionManager);
+ prepareAndReceiveInitProducerId(producerId, Errors.NONE);
+ assertTrue(transactionManager.hasProducerId());
+ assertEquals(0, transactionManager.sequenceNumber(tp0));
+
+ // Send the first ProduceRequest with sequence 0. It is created 1000ms
before the others so that it expires
+ // first (deliveryTimeoutMs is 1500).
+ Future<RecordMetadata> request1 = appendToAccumulator(tp0);
+ sender.runOnce();
+
+ time.sleep(1000L);
+
+ // Send the second and third ProduceRequests with sequences 1 and 2.
+ Future<RecordMetadata> request2 = appendToAccumulator(tp0);
+ sender.runOnce();
+ Future<RecordMetadata> request3 = appendToAccumulator(tp0);
+ sender.runOnce();
+ assertEquals(3, client.inFlightRequestCount());
+
+ // The first request fails because the partition completion has not
completed on the leader broker yet.
+ sendIdempotentProducerResponse(0, tp0, Errors.NOT_LEADER_OR_FOLLOWER,
-1L);
+ sender.runOnce(); // receive response 0
+
+ // The partition creation completes afterwards, so the request with
sequence 0 does not reach the partition
+ // and the broker rejects the in-flight second and third requests.
They are retried without an epoch bump
+ // since the retry of th first request could still fill the gap in the
expected sequence.
Review Comment:
typo th
--
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]
