[MNT-25691] Regression fix where the transactions are stalled (#2255)

2026-09-16 18:12:56 +00:00 · 2026-04-24 11:25:13 +05:30
parent 2be3c63a8b
commit 20e096cca6
2 changed files with 107 additions and 5 deletions
@@ -96,6 +96,18 @@ public class MetadataTracker extends ActivatableTracker
    private final ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<>();
    private ForkJoinPool forkJoinPool;

+    // Counts how many cycles ended with non-empty transactions all already indexed.
+    // After ALL_INDEXED_CYCLE_COUNT_THRESHOLD, getTxFromCommitTime() uses
+    // lastProcessedTxCommitTime to skip past the stale block.
+    // Reset when getTxFromCommitTime() sees non-empty txnsFound, or on invalidateState().
+    private volatile int allIndexedCycleCount = 0;
+
+    private static final int ALL_INDEXED_CYCLE_COUNT_THRESHOLD = 3;
+
+    // Highest commit time we saw in an all-already-indexed batch.
+    // Used in getTxFromCommitTime() once the threshold is reached.
+    private volatile long lastProcessedTxCommitTime = 0;
+
    // Share run and write locks across all MetadataTracker threads
    private static final Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
    private static final Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@@ -729,8 +741,16 @@ public class MetadataTracker extends ActivatableTracker
    protected Long getTxFromCommitTime(BoundedDeque<Transaction> txnsFound, long lastGoodTxCommitTimeInIndex) {
        if (txnsFound.size() > 0)
        {
+            // Previous do-while iteration indexed real transactions.
+            // Reset stuck counter so the bookmark doesn't activate prematurely.
+            allIndexedCycleCount = 0;
            return txnsFound.getLast().getCommitTimeMs();
        }
+        else if (allIndexedCycleCount >= ALL_INDEXED_CYCLE_COUNT_THRESHOLD
+                && lastProcessedTxCommitTime > lastGoodTxCommitTimeInIndex)
+        {
+            return lastProcessedTxCommitTime;
+        }
        else
        {
            return lastGoodTxCommitTimeInIndex;
@@ -1019,11 +1039,18 @@ public class MetadataTracker extends ActivatableTracker
                        }
                        reachedLagBoundary = true;
                    }
-                    else
+                    else if (!transactions.getTransactions().isEmpty())
                    {
-                        // Nothing left to index in this cycle — all fetched transactions are already indexed.
-                        // Advance the tracker state and stop.
+                        // Everything was already indexed. Save a bookmark and count the cycle.
+                        // After enough consecutive stuck cycles, getTxFromCommitTime() will
+                        // use the bookmark to skip past this block.
+                        long maxBatchCommitTime = transactions.getTransactions().stream()
+                                .mapToLong(Transaction::getCommitTimeMs)
+                                .max()
+                                .orElse(lastProcessedTxCommitTime);
                        setLastTxCommitTimeAndTxIdInTrackerState(transactions);
+                        lastProcessedTxCommitTime = Math.max(lastProcessedTxCommitTime, maxBatchCommitTime);
+                        allIndexedCycleCount++;
                        break;
                    }
                }
@@ -1426,6 +1453,8 @@ public class MetadataTracker extends ActivatableTracker

    public void invalidateState() {
        super.invalidateState();
+        lastProcessedTxCommitTime = 0;
+        allIndexedCycleCount = 0;
        infoSrv.clearProcessedTransactions();
    }

@@ -418,4 +418,77 @@ public class MetadataTrackerTest
        verify(srv, never()).continueState(any());
        assertEquals(1L, state.getTrackerCycles());
    }
+
+    /**
+     * Verifies that the tracker skips past a block of already-indexed transactions
+     * after ALL_INDEXED_CYCLES_THRESHOLD consecutive stuck cycles.
+     * Cycles 1-3 behave identically to master (break immediately).
+     * Cycle 4 uses the bookmark to start scanning past the stale block.
+     */
+    @Test
+    public void trackerAdvancesPastIndexedPage() throws Exception
+    {
+        TrackerState state = new TrackerState();
+        state.setTimeToStopIndexing(5000L);
+        state.setLastIndexedTxCommitTime(200L);
+        state.setLastIndexedTxId(2L);
+        // lastGoodTxCommitTimeInIndex defaults to 0 (simulates hole retention lag)
+
+        metadataTracker.state = state;
+        when(metadataTracker.getTrackerState()).thenReturn(state);
+
+        // Batch 1: two transactions that are already indexed
+        Transaction alreadyIndexed1 = new Transaction();
+        alreadyIndexed1.setId(1L);
+        alreadyIndexed1.setCommitTimeMs(100L);
+        alreadyIndexed1.setUpdates(1);
+
+        Transaction alreadyIndexed2 = new Transaction();
+        alreadyIndexed2.setId(2L);
+        alreadyIndexed2.setCommitTimeMs(200L);
+        alreadyIndexed2.setUpdates(1);
+
+        // Batch 2: one new transaction that needs indexing
+        Transaction newTx = new Transaction();
+        newTx.setId(3L);
+        newTx.setCommitTimeMs(300L);
+        newTx.setUpdates(1);
+
+        Transactions batch1 = new Transactions(List.of(alreadyIndexed1, alreadyIndexed2));
+        Transactions batch2 = new Transactions(List.of(newTx));
+
+        // Marking the existing transactions as already in the index
+        when(srv.txnInIndex(1L, true)).thenReturn(true);
+        when(srv.txnInIndex(2L, true)).thenReturn(true);
+
+        // Cycles 1-3: same batch each time, all already indexed -> bookmark + break
+        // Cycle 4: threshold hit, bookmark skips to 200, gets batch2 with new tx
+        when(repositoryClient.getTransactions(anyLong(), isNull(), anyLong(), isNull(), anyInt()))
+            .thenReturn(batch1)   // cycle 1
+            .thenReturn(batch1)   // cycle 2
+            .thenReturn(batch1)   // cycle 3
+            .thenReturn(batch2)   // cycle 4
+            .thenReturn(new Transactions(Collections.emptyList()));
+
+        // Cycles 1-3: identical to master
+        metadataTracker.trackTransactions();
+        metadataTracker.trackTransactions();
+        metadataTracker.trackTransactions();
+
+        // The already-indexed transactions must not be re-indexed
+        verify(srv, never()).indexTransaction(eq(alreadyIndexed1), anyBoolean());
+        verify(srv, never()).indexTransaction(eq(alreadyIndexed2), anyBoolean());
+        // Not yet reached the new transaction
+        verify(srv, never()).indexTransaction(eq(newTx), anyBoolean());
+
+        // Cycle 4: threshold reached, bookmark skips past stale block
+        metadataTracker.trackTransactions();
+
+        // New transaction gets indexed
+        verify(srv, times(1)).indexTransaction(eq(newTx), eq(true));
+        verify(trackerStats, times(1)).addTxDocs(1);
+
+        assertEquals(300L, state.getLastIndexedTxCommitTime());
+        assertEquals(3L, state.getLastIndexedTxId());
+    }
 }