Fix subscription AT_END replay cursor boundary

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java

@@ -1103,11 +1103,17 @@ && compareWriterProgress(requestProgress, storedWriterProgress) == 0) {
           "resolved first uncovered replayable WAL record");
     }
     return ReplayLocateDecision.atEnd(
-        consensusReqReader.getCurrentSearchIndex(),
+        nextSearchIndexAfterCurrent(),
         effectiveRecoveryRegionProgress,
         "all locally replayable WAL records are already covered");
   }
 
+  private long nextSearchIndexAfterCurrent() {
+    // The reader reports the last local WAL index, while the subscription cursor is the next
+    // index to read. Starting at the last index would replay it and report one spurious WAL gap.
+    return consensusReqReader.getCurrentSearchIndex() + 1L;
+  }
+
   protected ReplayLocateDecision locateReplayStartForRegionProgress(
       final RegionProgress regionProgress, final boolean seekAfter) {
     if (!(consensusReqReader instanceof WALNode)) {
@@ -3079,8 +3085,7 @@ public void seekToBeginning() {
    * Seeks to the current WAL write position. After this, only newly written data will be consumed.
    */
   public void seekToEnd() {
-    seekToResolvedPosition(
-        consensusReqReader.getCurrentSearchIndex(), computeTailRegionProgress(), "end");
+    seekToResolvedPosition(nextSearchIndexAfterCurrent(), computeTailRegionProgress(), "end");
   }
 
   public void seekToRegionProgress(final RegionProgress regionProgress) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java

@@ -612,12 +612,13 @@ public void testReplayStartPreservesUncoveredFollowerEntries() throws Exception
       reader.currentSearchIndex = 5L;
       final ConsensusPrefetchingQueue.ReplayLocateDecision tailDecision =
           queue.scanReplayStartForRequests(
-              Collections.singletonList(createRequest(-1L, 11L, 101L, 8)).iterator(),
+              Arrays.asList(createRequest(5L, 10L, 100L, 8), createRequest(-1L, 11L, 101L, 8))
+                  .iterator(),
               regionProgress,
               true);
 
       assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.AT_END, tailDecision.getStatus());
-      assertEquals(5L, tailDecision.getStartSearchIndex());
+      assertEquals(6L, tailDecision.getStartSearchIndex());
       assertEquals(
           committedProgress,
           tailDecision.getRecoveryRegionProgress().getWriterPositions().get(formerLeader));
@@ -1129,6 +1130,7 @@ public void testAtEndReplayLookupPreservesRequestedWriterFrontier() throws Excep
           queue.scanReplayStartForRequests(Collections.emptyIterator(), requestedProgress, true);
 
       assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.AT_END, decision.getStatus());
+      assertEquals(6L, decision.getStartSearchIndex());
       assertEquals(requestedProgress, decision.getRecoveryRegionProgress());
     } finally {
       if (queue != null) {
@@ -1138,6 +1140,53 @@ public void testAtEndReplayLookupPreservesRequestedWriterFrontier() throws Excep
     }
   }
 
+  @Test
+  public void testAtEndInitializationHasNoTailGap() throws Exception {
+    final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+    final File systemDir = temporaryFolder.newFolder("atEndNoTailGap");
+    final File walDirectory = temporaryFolder.newFolder("atEndNoTailGapWal");
+    ConsensusPrefetchingQueue queue = null;
+    try {
+      final DataRegionId regionId = new DataRegionId(7);
+      final WALNode walNode = mock(WALNode.class);
+      when(walNode.getLogDirectory()).thenReturn(walDirectory);
+      when(walNode.getCurrentSearchIndex()).thenReturn(5L);
+      final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class);
+      when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+      when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
+      final RegionProgress committedProgress =
+          new RegionProgress(
+              Collections.singletonMap(
+                  new WriterId(regionId.toString(), 7), new WriterProgress(1000L, 5L)));
+      queue =
+          new ConsensusPrefetchingQueue(
+              "consumerGroup",
+              "topic",
+              TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+              regionId,
+              serverImpl,
+              new SubscriptionWalRetentionPolicy(
+                  "topic",
+                  SubscriptionWalRetentionPolicy.UNBOUNDED,
+                  SubscriptionWalRetentionPolicy.UNBOUNDED),
+              mock(ConsensusLogToTabletConverter.class),
+              newCommitManager(systemDir),
+              committedProgress,
+              6L,
+              1L,
+              true);
+
+      assertNull(queue.poll("consumer"));
+      assertEquals(6L, queue.getCurrentReadSearchIndex());
+      assertEquals(0L, queue.getRawWalGap());
+    } finally {
+      if (queue != null) {
+        queue.close();
+      }
+      IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+    }
+  }
+
   @Test
   public void testAtEndInitializationPreservesPendingQueueRegistrationBoundary() throws Exception {
     final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();