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 500f70faa..09298906c 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 @@ -96,6 +96,18 @@ public class MetadataTracker extends ActivatableTracker private final ConcurrentLinkedQueue 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 RUN_LOCK_BY_CORE = new ConcurrentHashMap<>(); private static final Map WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>(); @@ -113,7 +125,7 @@ public class MetadataTracker extends ActivatableTracker /** * Check if nextTxCommitTimeService is available in the repository. * This service is used to find the next available transaction commit time from a given time, - * so periods of time where no document updating is happening can be skipped while getting + * so periods of time where no document updating is happening can be skipped while getting * pending transactions list. * * {@link org.alfresco.solr.client.SOLRAPIClient#GET_NEXT_TX_COMMIT_TIME} @@ -132,7 +144,7 @@ public class MetadataTracker extends ActivatableTracker private boolean cascadeTrackerEnabled = true; /** - * Transaction Id range to get the first transaction in database. + * Transaction Id range to get the first transaction in database. * 0-2000 by default. */ private Pair minTxnIdRange; @@ -729,8 +741,16 @@ public class MetadataTracker extends ActivatableTracker protected Long getTxFromCommitTime(BoundedDeque 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(); } 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 8f623f992..d27fa6ffa 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 @@ -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()); + } }