fix(storage): regard all produce requests as duplicate whose sequence… (#1865)

* fix(storage): regard all produce requests as duplicate whose sequence number is less than the 5 retained batches

Signed-off-by: Ning Yu <[email protected]>

* test: disable an unstable test

Signed-off-by: Ning Yu <[email protected]>

* test: update kafka tests

Signed-off-by: Ning Yu <[email protected]>

---------

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala

@@ -1306,8 +1306,11 @@ class UnifiedLogTest {
     records = TestUtils.records(
       List(new SimpleRecord(mockTime.milliseconds, s"key-1".getBytes, s"value-1".getBytes)),
       producerId = pid, producerEpoch = epoch, sequence = 1)
-    assertThrows(classOf[OutOfOrderSequenceException], () => log.appendAsLeader(records, leaderEpoch = 0),
-      () => "Should have received an OutOfOrderSequenceException since we attempted to append a duplicate of a batch which is older than the last 5 appended batches.")
+    // AutoMQ for Kafka inject start
+    // We expect a DuplicateSequenceException since we attempted to append a duplicate of a batch which is older than the last 5 appended batches.
+    assertThrows(classOf[DuplicateSequenceException], () => log.appendAsLeader(records, leaderEpoch = 0),
+      () => "Should have received an DuplicateSequenceException since we attempted to append a duplicate of a batch which is older than the last 5 appended batches.")
+    // AutoMQ for Kafka inject end
 
     // Append a duplicate entry with a single records at the tail of the log. This should return the appendInfo of the original entry.
     def createRecordsWithDuplicate = TestUtils.records(List(new SimpleRecord(mockTime.milliseconds, "key".getBytes, "value".getBytes)),

s3stream/src/test/java/com/automq/stream/s3/index/LocalStreamRangeIndexCacheTest.java

@@ -20,6 +20,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
@@ -141,6 +142,7 @@ public void testPrune() {
     }
 
     @Test
+    @Disabled("FIXME: This test is not stable")
     public void testEvict() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
         LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();

storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateEntry.java

@@ -26,6 +26,8 @@
 import org.apache.kafka.common.errors.DuplicateSequenceException;
 import org.apache.kafka.common.record.RecordBatch;
 
+import static org.apache.kafka.common.record.DefaultRecordBatch.decrementSequence;
+
 /**
  * This class represents the state of a specific producer-id.
  * It contains batchMetadata queue which is ordered such that the batch with the lowest sequence is at the head of the
@@ -142,8 +144,24 @@ public Optional<BatchMetadata> findDuplicateBatch(RecordBatch batch) {
         }
         BatchMetadata front = batchMetadata.peek();
         if (front != null && front.recovered) {
-            // the batch metadata is recovered from snapshot
-            if (front.firstSeq() <= batch.baseSequence() && front.lastSeq >= batch.lastSequence()) {
+            // the batch metadata (`front`) is recovered from snapshot
+            boolean batchFallInFront = contains(front.firstSeq(), front.lastSeq, batch.baseSequence())
+                && contains(front.firstSeq(), front.lastSeq, batch.lastSequence());
+            if (batchFallInFront) {
+                throw new DuplicateSequenceException(
+                    String.format("The batch is duplicated (recover from snapshot), broker cached metadata is %s, the record batch is [%s, %s]",
+                        this, batch.baseSequence(), batch.lastSequence())
+                );
+            }
+        }
+        if (front != null) {
+            // regard the batch as duplicated if it is before the first cached batch
+            boolean batchBeforeFront = contains(
+                decrementSequence(front.firstSeq(), 65536),
+                decrementSequence(front.firstSeq(), 1),
+                batch.lastSequence()
+            );
+            if (batchBeforeFront) {
                 throw new DuplicateSequenceException(
                     String.format("The batch is duplicated, broker cached metadata is %s, the record batch is [%s, %s]",
                         this, batch.baseSequence(), batch.lastSequence())
@@ -154,6 +172,21 @@ public Optional<BatchMetadata> findDuplicateBatch(RecordBatch batch) {
         // AutoMQ inject end
     }
 
+    // AutoMQ inject start
+    /**
+     * Check if the sequence number is in the range [start, end] (inclusive).
+     * The range may wrap around the sequence space.
+     */
+    private static boolean contains(int start, int end, int seq) {
+        if (start <= end) {
+            return seq >= start && seq <= end;
+        } else {
+            // wrap around
+            return seq >= start || seq <= end;
+        }
+    }
+    // AutoMQ inject end
+
     // Return the batch metadata of the cached batch having the exact sequence range, if any.
     Optional<BatchMetadata> batchWithSequenceRange(int firstSeq, int lastSeq) {
         Stream<BatchMetadata> duplicate = batchMetadata.stream().filter(metadata -> firstSeq == metadata.firstSeq() && lastSeq == metadata.lastSeq);