From b0af50bc6fcd788efcd85fbe049b798f9feb0e2b Mon Sep 17 00:00:00 2001 From: tathagta15 Date: Fri, 20 Feb 2026 12:25:32 +0530 Subject: [PATCH] [MNT-25456] Introduced lag safe indexing logic (#2233) --- .../solr/tracker/MetadataTracker.java | 93 +++++++++++++++---- .../solr/tracker/MetadataTrackerTest.java | 91 ++++++++++++++++++ 2 files changed, 167 insertions(+), 17 deletions(-) diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java index 8ef7e5d2f..500f70faa 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java @@ -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 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> txBatches = transactions.getTransactions().stream() - .peek(txnsFound::add) + List lagEligibleTransactions = transactions.getTransactions().stream() + .filter(transaction -> transaction.getCommitTimeMs() <= lagCutoff) .filter(this::isTransactionToBeIndexed) + .peek(txnsFound::add) + .collect(Collectors.toList()); + + hitLagBoundary.set(!deferredTransactions.isEmpty()); + + Collection> 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> nodeBatches = new ArrayList<>(); - for (List batch : txBatches) + for (List 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 batch : txBatches) + for (List 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 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); } diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java index bb9e61f79..db8dadf36 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java @@ -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();