Remove unproven seek boundary changes from subscription fix

Author: Caideyipi

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

@@ -882,9 +882,7 @@ private ReplayLocateDecision resolveInitReplayStartDecision(
       case AT_END:
         final long resolvedReplayStartSearchIndex =
             replayTarget.getStatus() == ReplayLocateStatus.AT_END
-                ? Math.min(
-                    nextSearchIndexAfter(replayTarget.getStartSearchIndex()),
-                    fallbackTailSearchIndex)
+                ? Math.min(replayTarget.getStartSearchIndex(), fallbackTailSearchIndex)
                 : replayTarget.getStartSearchIndex();
         return new ReplayLocateDecision(
             replayTarget.getStatus(),
@@ -2037,6 +2035,7 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque
     if (!hasLocalSearchIndex(request)) {
       return;
     }
+
     final long actualSearchIndex = request.getSearchIndex();
     final long expectedSearchIndex = nextExpectedSearchIndex.get();
     if (actualSearchIndex > expectedSearchIndex) {
@@ -3073,28 +3072,15 @@ public void cleanUp() {
    * old files have been reclaimed, the earliest available position may be later than 0.
    */
   public void seekToBeginning() {
-    final RegionProgress beginningProgress = new RegionProgress(Collections.emptyMap());
-    if (consensusReqReader instanceof WALNode) {
-      // The first retained WAL index can be greater than one after older files are reclaimed.
-      seekToRegionProgress(beginningProgress);
-    } else {
-      seekToResolvedPosition(FIRST_CONSENSUS_SEARCH_INDEX, beginningProgress, "beginning");
-    }
+    seekToResolvedPosition(0L, new RegionProgress(Collections.emptyMap()), "beginning");
   }
 
   /**
    * Seeks to the current WAL write position. After this, only newly written data will be consumed.
    */
   public void seekToEnd() {
-    final RegionProgress tailProgress = computeTailRegionProgress();
     seekToResolvedPosition(
-        nextSearchIndexAfter(consensusReqReader.getCurrentSearchIndex()), tailProgress, "end");
-  }
-
-  private static long nextSearchIndexAfter(final long searchIndex) {
-    return searchIndex == Long.MAX_VALUE
-        ? Long.MAX_VALUE
-        : Math.max(FIRST_CONSENSUS_SEARCH_INDEX, searchIndex + 1L);
+        consensusReqReader.getCurrentSearchIndex(), computeTailRegionProgress(), "end");
   }
 
   public void seekToRegionProgress(final RegionProgress regionProgress) {
@@ -3122,9 +3108,7 @@ public void seekToRegionProgress(final RegionProgress regionProgress) {
             replayTarget.getStatus(),
             replayTarget.getStartSearchIndex());
         seekToResolvedPosition(
-            replayTarget.getStatus() == ReplayLocateStatus.AT_END
-                ? nextSearchIndexAfter(replayTarget.getStartSearchIndex())
-                : replayTarget.getStartSearchIndex(),
+            replayTarget.getStartSearchIndex(),
             replayTarget.getRecoveryRegionProgress(),
             "regionProgress");
         return;
@@ -3165,9 +3149,7 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) {
             replayTarget.getStatus(),
             replayTarget.getStartSearchIndex());
         seekToResolvedPosition(
-            replayTarget.getStatus() == ReplayLocateStatus.AT_END
-                ? nextSearchIndexAfter(replayTarget.getStartSearchIndex())
-                : replayTarget.getStartSearchIndex(),
+            replayTarget.getStartSearchIndex(),
             replayTarget.getRecoveryRegionProgress(),
             "regionProgressAfter");
         return;

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

@@ -1210,67 +1210,6 @@ protected ReplayLocateDecision locateReplayStartForRegionProgress(
     }
   }
 
-  @Test
-  public void testAtEndInitializationStartsAfterLastWalIndex() throws Exception {
-    final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
-    final File systemDir = temporaryFolder.newFolder("atEndAfterTail");
-    final File walDirectory = temporaryFolder.newFolder("atEndAfterTailWal");
-    ConsensusPrefetchingQueue queue = null;
-    try {
-      final DataRegionId regionId = new DataRegionId(8);
-      final WALNode walNode = mock(WALNode.class);
-      when(walNode.getLogDirectory()).thenReturn(walDirectory);
-      when(walNode.getCurrentSearchIndex()).thenReturn(4L);
-      final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class);
-      when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
-      when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
-      final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class);
-      when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet()));
-      when(converter.getDatabaseName()).thenReturn("db");
-      final RegionProgress committedProgress =
-          new RegionProgress(
-              Collections.singletonMap(
-                  new WriterId(regionId.toString(), 7), new WriterProgress(1000L, 2L)));
-      queue =
-          new ConsensusPrefetchingQueue(
-              "consumerGroup",
-              "topic",
-              TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
-              regionId,
-              serverImpl,
-              new SubscriptionWalRetentionPolicy(
-                  "topic",
-                  SubscriptionWalRetentionPolicy.UNBOUNDED,
-                  SubscriptionWalRetentionPolicy.UNBOUNDED),
-              converter,
-              newCommitManager(systemDir),
-              committedProgress,
-              5L,
-              1L,
-              true) {
-            @Override
-            protected ReplayLocateDecision locateReplayStartForRegionProgress(
-                final RegionProgress regionProgress, final boolean seekAfter) {
-              return ReplayLocateDecision.atEnd(
-                  walNode.getCurrentSearchIndex(), regionProgress, "empty sealed WAL");
-            }
-          };
-      queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024));
-
-      assertNull(queue.poll("consumer"));
-      assertEquals(5L, queue.getCurrentReadSearchIndex());
-      assertTrue(pendingEntries(queue).offer(createRequest(5L)));
-      queue.drivePrefetchOnce();
-      assertEquals(6L, queue.getCurrentReadSearchIndex());
-      assertEquals(1L, queue.getPendingPathAcceptedEntries());
-    } finally {
-      if (queue != null) {
-        queue.close();
-      }
-      IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
-    }
-  }
-
   @Test
   public void testZeroPhysicalTimeUsesWriterLocalSequenceForReplayLookup() throws Exception {
     final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();