[MNT-25456] Introduced lag safe indexing logic (#2233)

2026-09-16 18:12:56 +00:00 · 2026-02-20 12:25:32 +05:30
parent e9b6f7fd40
commit b0af50bc6f
2 changed files with 167 additions and 17 deletions
@@ -59,6 +59,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;

@@ -918,6 +919,7 @@ public class MetadataTracker extends ActivatableTracker
        int totalUpdatedDocs = 0;

        LOGGER.info("{}-[CORE {}] Starting metadata tracker execution", Thread.currentThread().getId(), coreName);
+        boolean reachedLagBoundary = false;

        do
        {
@@ -950,6 +952,8 @@ public class MetadataTracker extends ActivatableTracker
                }

                long idTrackerCycle = System.currentTimeMillis();
+                long lagCutoff = state.getTimeToStopIndexing();
+                AtomicBoolean hitLagBoundary = new AtomicBoolean(false);
                if (transactions.getTransactions().size() > 0)
                {
                    LOGGER.info("{}:{}-[CORE {}] Found {} transactions after lastTxCommitTime {}, transactions from {} to {}",
@@ -971,26 +975,62 @@ public class MetadataTracker extends ActivatableTracker
                                    : state.getLastIndexedTxCommitTime()));
                }

-                // Make sure we do not go ahead of where we started - we will check the holes here
-                // correctly next time
-                if (transactions.getTransactions()
-                        .stream()
-                        .anyMatch(transaction -> transaction.getCommitTimeMs() > state.getTimeToStopIndexing()))
+                final AtomicInteger counterTransaction = new AtomicInteger();
+
+                List<Transaction> deferredTransactions = transactions.getTransactions().stream()
+                        .filter(transaction -> transaction.getCommitTimeMs() > lagCutoff)
+                        .collect(Collectors.toList());
+
+                if (!deferredTransactions.isEmpty() && LOGGER.isDebugEnabled())
                {
-                    break;
+                    deferredTransactions.forEach(transaction -> LOGGER.debug("{}:{}-[CORE {}] Deferring transaction {} at {} past lag cutoff {}",
+                            Thread.currentThread().getId(),
+                            idTrackerCycle,
+                            coreName,
+                            transaction.getId(),
+                            transaction.getCommitTimeMs(),
+                            lagCutoff));
                }

-                final AtomicInteger counterTransaction = new AtomicInteger();
-                Collection<List<Transaction>> txBatches = transactions.getTransactions().stream()
-                        .peek(txnsFound::add)
+                List<Transaction> lagEligibleTransactions = transactions.getTransactions().stream()
+                        .filter(transaction -> transaction.getCommitTimeMs() <= lagCutoff)
                        .filter(this::isTransactionToBeIndexed)
+                        .peek(txnsFound::add)
+                        .collect(Collectors.toList());
+
+                hitLagBoundary.set(!deferredTransactions.isEmpty());
+
+                Collection<List<Transaction>> eligibleTransactionBatches = lagEligibleTransactions.stream()
                        .collect(Collectors.groupingBy(transaction -> counterTransaction.getAndAdd(
                                (int) (transaction.getDeletes() + transaction.getUpdates())) / transactionDocsBatchSize))
                        .values();

+                if (eligibleTransactionBatches.isEmpty())
+                {
+                    if (hitLagBoundary.get())
+                    {
+                        if(LOGGER.isDebugEnabled())
+                        {
+                            LOGGER.debug("{}:{}-[CORE {}] Reached lag cutoff {}, deferring newer transactions",
+                                    Thread.currentThread().getId(),
+                                    idTrackerCycle,
+                                    coreName,
+                                    lagCutoff);
+                        }
+                        reachedLagBoundary = true;
+                    }
+                    else
+                    {
+                        // Nothing left to index in this cycle — all fetched transactions are already indexed.
+                        // Advance the tracker state and stop.
+                        setLastTxCommitTimeAndTxIdInTrackerState(transactions);
+                        break;
+                    }
+                }
+
                // Index batches of transactions and the nodes updated or deleted within the transaction
                List<List<Node>> nodeBatches = new ArrayList<>();
-                for (List<Transaction> batch : txBatches)
+                for (List<Transaction> batch : eligibleTransactionBatches)
                {

                    // Index nodes contained in the transactions
@@ -1014,19 +1054,38 @@ public class MetadataTracker extends ActivatableTracker
                            return batch.size();
                        }).reduce(0, Integer::sum)).get();

-                for (List<Transaction> batch : txBatches)
+                for (List<Transaction> batch : eligibleTransactionBatches)
                {
-                    // Add the transactions as found to avoid processing them again in the next iteration
-                    batch.forEach(txnsFound::add);
-    
                    // Index the transactions
                    indexTransactionsAfterWorker(batch);
                    long endElapsed = System.nanoTime();
                    trackerStats.addElapsedNodeTime(totalUpdatedDocs, endElapsed - startElapsed);
                    startElapsed = endElapsed;
                }
-                
-                setLastTxCommitTimeAndTxIdInTrackerState(transactions);
+
+                // Set the tracker state only for transactions we actually indexed in this cycle
+                List<Transaction> indexedTransactions = eligibleTransactionBatches.stream()
+                    .flatMap(Collection::stream)
+                    .collect(Collectors.toList());
+                if (!indexedTransactions.isEmpty())
+                {
+                    long maxIndexedCommitTime = indexedTransactions.stream()
+                        .mapToLong(Transaction::getCommitTimeMs)
+                        .max()
+                        .orElse(state.getLastTxCommitTimeOnServer());
+
+                    long maxIndexedTxId = indexedTransactions.stream()
+                        .mapToLong(Transaction::getId)
+                        .max()
+                        .orElse(state.getLastTxIdOnServer());
+
+                    setLastTxCommitTimeAndTxIdInTrackerState(new Transactions(indexedTransactions, maxIndexedCommitTime, maxIndexedTxId));
+                }
+
+                if (hitLagBoundary.get())
+                {
+                    reachedLagBoundary = true;
+                }
            }
            catch(Exception e)
            {
@@ -1038,7 +1097,7 @@ public class MetadataTracker extends ActivatableTracker
            }
        
        }
-        while ((transactions.getTransactions().size() > 0));
+        while (!reachedLagBoundary && (transactions.getTransactions().size() > 0));

        LOGGER.info("{}-[CORE {}] Tracked {} DOCs", Thread.currentThread().getId(), coreName, totalUpdatedDocs);
    }
@@ -28,6 +28,7 @@ package org.alfresco.solr.tracker;

 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Properties;

@@ -58,6 +59,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
@@ -277,6 +279,95 @@ public class MetadataTrackerTest
        assertTrue(metadataTracker.isTransactionToBeIndexed(incomingTransaction));
    }

+    @Test
+    public void lastIndexedTransactionUsesCommitTimeThenIdOrdering() throws Exception
+    {
+        TrackerState state = new TrackerState();
+        state.setTimeToStopIndexing(1000L);
+        state.setLastIndexedTxCommitTime(0L);
+        state.setLastIndexedTxId(0L);
+
+        metadataTracker.state = state;
+        when(metadataTracker.getTrackerState()).thenReturn(state);
+
+        Transaction tx1 = new Transaction();
+        tx1.setId(1L);
+        tx1.setCommitTimeMs(100L);
+        tx1.setUpdates(1);
+
+        Transaction tx2 = new Transaction();
+        tx2.setId(2L);
+        tx2.setCommitTimeMs(100L);
+        tx2.setUpdates(1);
+
+        Transaction tx3 = new Transaction();
+        tx3.setId(3L);
+        tx3.setCommitTimeMs(90L);
+        tx3.setUpdates(1);
+
+        Transactions transactions = new Transactions(List.of(tx1, tx3, tx2));
+        when(repositoryClient.getTransactions(anyLong(), isNull(), anyLong(), isNull(), anyInt()))
+            .thenReturn(transactions)
+            .thenReturn(new Transactions(Collections.emptyList()));
+
+        metadataTracker.trackTransactions();
+
+        assertEquals(100L, state.getLastIndexedTxCommitTime());
+        assertEquals(2L, state.getLastIndexedTxId());
+        // Verify all eligible transactions were indexed by the tracker
+        verify(srv, times(1)).indexTransaction(eq(tx1), eq(true));
+        verify(srv, times(1)).indexTransaction(eq(tx2), eq(true));
+        // Ordering is not asserted here; only eligibility and lastIndexed* logic are.
+        verify(srv, times(1)).indexTransaction(eq(tx3), eq(true));
+        verify(trackerStats, times(3)).addTxDocs(1);
+    }
+
+    @Test
+    public void lagCutoffFiltersDeferredTransactionsBeforeBatching() throws Exception
+    {
+        TrackerState state = new TrackerState();
+        state.setTimeToStopIndexing(100L);
+        state.setLastIndexedTxCommitTime(0L);
+        state.setLastIndexedTxId(0L);
+
+        metadataTracker.state = state;
+        when(metadataTracker.getTrackerState()).thenReturn(state);
+
+        Transaction oldTx = new Transaction();
+        oldTx.setId(10L);
+        oldTx.setCommitTimeMs(90L);
+        oldTx.setUpdates(1);
+
+        Transaction boundaryTx = new Transaction();
+        boundaryTx.setId(11L);
+        boundaryTx.setCommitTimeMs(100L);
+        boundaryTx.setUpdates(1);
+
+        Transaction newTx = new Transaction();
+        newTx.setId(12L);
+        newTx.setCommitTimeMs(110L);
+        newTx.setUpdates(1);
+
+        Transactions transactions = new Transactions(List.of(oldTx, boundaryTx, newTx));
+        when(repositoryClient.getTransactions(anyLong(), isNull(), anyLong(), isNull(), anyInt()))
+            .thenReturn(transactions)
+            .thenReturn(new Transactions(Collections.emptyList()));
+
+        metadataTracker.trackTransactions();
+
+        // Verify only lag-eligible transactions were indexed
+        verify(srv, times(1)).indexTransaction(eq(oldTx), eq(true));
+        verify(srv, times(1)).indexTransaction(eq(boundaryTx), eq(true));
+        verify(srv, never()).indexTransaction(eq(newTx), anyBoolean());
+
+        // Verify we stop after hitting the lag boundary
+        verify(repositoryClient, times(1)).getTransactions(anyLong(), isNull(), anyLong(), isNull(), anyInt());
+        verify(trackerStats, times(2)).addTxDocs(1);
+
+        assertEquals(100L, state.getLastIndexedTxCommitTime());
+        assertEquals(11L, state.getLastIndexedTxId());
+    }
+
    private Node getNode()
    {
        Node node = new Node();