fix(wal): filter inner trim record for range get (#3000)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

s3stream/src/main/java/com/automq/stream/s3/wal/impl/object/DefaultReader.java

@@ -47,6 +47,7 @@
 import static com.automq.stream.s3.wal.impl.object.ObjectUtils.floorAlignOffset;
 import static com.automq.stream.s3.wal.impl.object.ObjectUtils.genObjectPathV1;
 
+@SuppressWarnings("checkstyle:cyclomaticComplexity")
 @EventLoopSafe
 public class DefaultReader {
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultReader.class);
@@ -190,7 +191,11 @@ private void doRunBatchGet0(BatchReadTask readTask) {
                                 long nextRecordOffset = finalNextGetOffset;
                                 int lastReadableBytes = buf.readableBytes();
                                 while (buf.readableBytes() > 0 && nextRecordOffset < readTask.endOffset.offset()) {
-                                    batches.add(ObjectUtils.decodeRecordBuf(buf));
+                                    StreamRecordBatch batch = ObjectUtils.decodeRecordBuf(buf);
+                                    boolean isTriggerTrimRecord = batch.getCount() == 0 && batch.getStreamId() == -1L && batch.getEpoch() == -1L;
+                                    if (!isTriggerTrimRecord) {
+                                        batches.add(batch);
+                                    }
                                     nextRecordOffset += lastReadableBytes - buf.readableBytes();
                                     lastReadableBytes = buf.readableBytes();
                                 }

s3stream/src/test/java/com/automq/stream/s3/wal/impl/object/ObjectWALServiceTest.java

@@ -119,6 +119,44 @@ public void testGet_batch() throws Exception {
         }
     }
 
+    @Test
+    public void testGet_batchSkipTrim() throws Exception {
+        ObjectWALConfig config = ObjectWALConfig.builder()
+            .withEpoch(1L)
+            .withMaxBytesInBatch(1024)
+            .withBatchInterval(1000)
+            .build();
+        ObjectWALService wal = new ObjectWALService(time, objectStorage, config);
+        acquire(config);
+        wal.start();
+
+        List<CompletableFuture<AppendResult>> appendCfList = new ArrayList<>();
+        for (int i = 0; i < 8; i++) {
+            appendCfList.add(wal.append(TraceContext.DEFAULT,
+                new StreamRecordBatch(233L, 10, 100L + i, 1, generateByteBuf(256))));
+            // ensure objects are flushed/uploaded
+            ((DefaultWriter) (wal.writer)).flush().join();
+            if (i == 4) {
+                // write a trim marker by trimming to the 4th record's offset (this will produce a trim record object)
+                wal.trim(appendCfList.get(3).get().recordOffset()).get();
+            }
+        }
+
+
+        // query across a range that spans objects including the trim marker
+        List<StreamRecordBatch> records = wal.get(
+            DefaultRecordOffset.of(appendCfList.get(4).get().recordOffset()),
+            DefaultRecordOffset.of(wal.confirmOffset())
+        ).get();
+
+        assertEquals(4, records.size());
+        for (int i = 0; i < records.size(); i++) {
+            assertEquals(104L + i, records.get(i).getBaseOffset());
+        }
+        wal.shutdownGracefully();
+    }
+
+
     @Test
     public void testTrim() throws Exception {
         ObjectWALConfig config = ObjectWALConfig.builder().withEpoch(1L).withMaxBytesInBatch(1024).withBatchInterval(1000).build();