junrao commented on code in PR #23234:
URL: https://github.com/apache/kafka/pull/23234#discussion_r3855738582
##########
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:
Is sleeping producerIdExpirationCheckIntervalMs enough?
producerIdExpirationMs is 200ms, twice of producerIdExpirationCheckIntervalMs.
--
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]
