From b530ef731cb095a245d025bef46020b2a5690a7f Mon Sep 17 00:00:00 2001 From: Elia Porciani Date: Wed, 3 Jun 2020 12:29:08 +0100 Subject: [PATCH] [SEARCH-2278] Modified DistributedAlfrescoSolrTrackerRace in order to test an hole of more than an hour between transactions --- .../solr/AlfrescoCoreAdminHandler.java | 4 +- .../alfresco/solr/HandlerReportHelper.java | 4 +- .../org/alfresco/solr/tracker/AclTracker.java | 226 ++++++----- .../alfresco/solr/tracker/CascadeTracker.java | 4 +- .../alfresco/solr/tracker/ContentTracker.java | 2 +- .../solr/tracker/MetadataTracker.java | 355 ++++++++---------- .../org/alfresco/solr/AlfrescoSolrUtils.java | 27 +- .../org/alfresco/solr/SolrITInitializer.java | 12 +- .../DistributedAlfrescoSolrTrackerRaceIT.java | 56 +-- ...rescoTrackerWithDelayedTransactionsIT.java | 169 +++++++++ .../tracker/DistributedCascadeTrackerIT.java | 6 +- 11 files changed, 506 insertions(+), 359 deletions(-) create mode 100644 search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoTrackerWithDelayedTransactionsIT.java diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java index 846b364ca..18ccb6d6f 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java @@ -1621,7 +1621,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler { // Gets Metadata health and fixes any problems MetadataTracker metadataTracker = trackerRegistry.getTrackerForCore(coreName, MetadataTracker.class); - IndexHealthReport indexHealthReport = metadataTracker.checkIndex(null, null, null, null); + IndexHealthReport indexHealthReport = metadataTracker.checkIndex(null, null, null); IOpenBitSet toReindex = indexHealthReport.getTxInIndexButNotInDb(); toReindex.or(indexHealthReport.getDuplicatedTxInIndex()); toReindex.or(indexHealthReport.getMissingTxFromIndex()); @@ -1636,7 +1636,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler // Gets the Acl health and fixes any problems AclTracker aclTracker = trackerRegistry.getTrackerForCore(coreName, AclTracker.class); - indexHealthReport = aclTracker.checkIndex(null, null, null, null); + indexHealthReport = aclTracker.checkIndex(null, null, null); toReindex = indexHealthReport.getAclTxInIndexButNotInDb(); toReindex.or(indexHealthReport.getDuplicatedAclTxInIndex()); toReindex.or(indexHealthReport.getMissingAclTxFromIndex()); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/HandlerReportHelper.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/HandlerReportHelper.java index a6738ba79..43d402de5 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/HandlerReportHelper.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/HandlerReportHelper.java @@ -157,7 +157,7 @@ class HandlerReportHelper { // ACL AclTracker aclTracker = trackerRegistry.getTrackerForCore(coreName, AclTracker.class); - IndexHealthReport aclReport = aclTracker.checkIndex(toTx, toAclTx, fromTime, toTime); + IndexHealthReport aclReport = aclTracker.checkIndex(toAclTx, fromTime, toTime); NamedList ihr = new SimpleOrderedMap<>(); ihr.add("DB acl transaction count", aclReport.getDbAclTransactionCount()); ihr.add("Count of duplicated acl transactions in the index", aclReport.getDuplicatedAclTxInIndex() @@ -187,7 +187,7 @@ class HandlerReportHelper // Metadata MetadataTracker metadataTracker = trackerRegistry.getTrackerForCore(coreName, MetadataTracker.class); - IndexHealthReport metaReport = metadataTracker.checkIndex(toTx, toAclTx, fromTime, toTime); + IndexHealthReport metaReport = metadataTracker.checkIndex(toTx, fromTime, toTime); ihr.add("DB transaction count", metaReport.getDbTransactionCount()); ihr.add("Count of duplicated transactions in the index", metaReport.getDuplicatedTxInIndex() .cardinality()); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java index e039893b6..bb40da9e0 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java @@ -56,8 +56,6 @@ import org.alfresco.solr.client.Acl; import org.alfresco.solr.client.AclChangeSet; import org.alfresco.solr.client.AclChangeSets; import org.alfresco.solr.client.AclReaders; -import org.alfresco.solr.client.GetNodesParameters; -import org.alfresco.solr.client.Node; import org.alfresco.solr.client.SOLRAPIClient; import org.json.JSONException; import org.slf4j.Logger; @@ -73,21 +71,30 @@ public class AclTracker extends AbstractTracker { protected final static Logger LOGGER = LoggerFactory.getLogger(AclTracker.class); + private static final int ACL_CHANGE_SETS_FOUND_QUEUE_SIZE = 100; private static final int DEFAULT_CHANGE_SET_ACLS_BATCH_SIZE = 2000; private static final int DEFAULT_ACL_BATCH_SIZE = 100; private static final int DEFAULT_ACL_TRACKER_MAX_PARALLELISM = 32; + private static final long DEFAULT_ACL_TRACKER_TIMESTEP = TIME_STEP_1_HR_IN_MS; + + private static final long INITIAL_MAX_ACL_CHANGE_SET_ID = 2000L; + private static final int MAX_NUMBER_OF_ACL_CHANGE_SETS = 2000; + + private static final long MAX_TIME_STEP = TIME_STEP_32_DAYS_IN_MS; private int aclTrackerParallelism; - private int changeSetAclsBatchSize = DEFAULT_CHANGE_SET_ACLS_BATCH_SIZE; - private int aclBatchSize = DEFAULT_ACL_BATCH_SIZE; + private int changeSetAclsBatchSize; + private int aclBatchSize; + private long timeStep; + private int maxNumberOfAclChangeSets; - private ConcurrentLinkedQueue aclChangeSetsToReindex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue aclChangeSetsToIndex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue aclChangeSetsToPurge = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue aclsToReindex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue aclsToIndex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue aclsToPurge = new ConcurrentLinkedQueue(); + private ConcurrentLinkedQueue aclChangeSetsToReindex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue aclChangeSetsToIndex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue aclChangeSetsToPurge = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue aclsToReindex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue aclsToIndex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue aclsToPurge = new ConcurrentLinkedQueue<>(); private DocRouter docRouter; private ForkJoinPool forkJoinPool; @@ -121,14 +128,21 @@ public class AclTracker extends AbstractTracker super(p, client, coreName, informationServer, Tracker.Type.ACL); changeSetAclsBatchSize = Integer.parseInt(p.getProperty("alfresco.changeSetAclsBatchSize", String.valueOf(DEFAULT_CHANGE_SET_ACLS_BATCH_SIZE))); - aclBatchSize = Integer.parseInt(p.getProperty("alfresco.aclBatchSize", String.valueOf(DEFAULT_ACL_BATCH_SIZE))); + aclBatchSize = Integer.parseInt(p.getProperty("alfresco.aclBatchSize", + String.valueOf(DEFAULT_ACL_BATCH_SIZE))); shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID); docRouter = DocRouterFactory.getRouter(p, ShardMethodEnum.getShardMethod(shardMethod)); - aclTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.aclTrackerMaxParallelism", + aclTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.acl.tracker.maxParallelism", String.valueOf(DEFAULT_ACL_TRACKER_MAX_PARALLELISM))); forkJoinPool = new ForkJoinPool(aclTrackerParallelism); + timeStep = Long.parseLong(p.getProperty("alfresco.acl.tracker.timestep", + String.valueOf(DEFAULT_ACL_TRACKER_TIMESTEP))); + + maxNumberOfAclChangeSets = Integer.parseInt(p.getProperty("alfresco.acl.tracker.maxNumberOfAclChangeSets", + String.valueOf(MAX_NUMBER_OF_ACL_CHANGE_SETS))); + RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true)); WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true)); } @@ -168,8 +182,10 @@ public class AclTracker extends AbstractTracker Long aclChangeSetId = aclChangeSetsToIndex.poll(); if (aclChangeSetId != null) { - AclChangeSets aclChangeSets = client.getAclChangeSets(null, aclChangeSetId, null, aclChangeSetId+1, 1); - if ((aclChangeSets.getAclChangeSets().size() > 0) && aclChangeSetId.equals(aclChangeSets.getAclChangeSets().get(0).getId())) + AclChangeSets aclChangeSets = client.getAclChangeSets(null, aclChangeSetId, + null, aclChangeSetId+1, 1); + if ((aclChangeSets.getAclChangeSets().size() > 0) && + aclChangeSetId.equals(aclChangeSets.getAclChangeSets().get(0).getId())) { AclChangeSet changeSet = aclChangeSets.getAclChangeSets().get(0); List acls = client.getAcls(Collections.singletonList(changeSet), null, Integer.MAX_VALUE); @@ -184,9 +200,8 @@ public class AclTracker extends AbstractTracker } else { - LOGGER.info( - "[CORE {}] - INDEX ACTION - AclChangeSetId {} was not found in database, it has NOT been reindexed", - coreName, aclChangeSetId); + LOGGER.info("[CORE {}] - INDEX ACTION - AclChangeSetId {} was not found in database, " + + "it has NOT been reindexed", coreName, aclChangeSetId); } } checkShutdown(); @@ -227,11 +242,14 @@ public class AclTracker extends AbstractTracker { this.infoSrv.deleteByAclChangeSetId(aclChangeSetId); - AclChangeSets aclChangeSets = client.getAclChangeSets(null, aclChangeSetId, null, aclChangeSetId+1, 1); - if ((aclChangeSets.getAclChangeSets().size() > 0) && aclChangeSetId.equals(aclChangeSets.getAclChangeSets().get(0).getId())) + AclChangeSets aclChangeSets = client.getAclChangeSets(null, aclChangeSetId, + null, aclChangeSetId+1, 1); + if ((aclChangeSets.getAclChangeSets().size() > 0) && + aclChangeSetId.equals(aclChangeSets.getAclChangeSets().get(0).getId())) { AclChangeSet changeSet = aclChangeSets.getAclChangeSets().get(0); - List acls = client.getAcls(Collections.singletonList(changeSet), null, Integer.MAX_VALUE); + List acls = client.getAcls(Collections.singletonList(changeSet), + null, Integer.MAX_VALUE); for (Acl acl : acls) { List readers = client.getAclReaders(Collections.singletonList(acl)); @@ -239,14 +257,14 @@ public class AclTracker extends AbstractTracker } this.infoSrv.indexAclTransaction(changeSet, true); - LOGGER.info("[CORE {}] - REINDEX ACTION - AclChangeSetId {} has been reindexed", coreName, aclChangeSetId); + LOGGER.info("[CORE {}] - REINDEX ACTION - AclChangeSetId {} has been reindexed", + coreName, aclChangeSetId); requiresCommit = true; } else { - LOGGER.info( - "[CORE {}] - REINDEX ACTION - AclChangeSetId {} was not found in database, it has NOT been reindexed", - coreName, aclChangeSetId); + LOGGER.info("[CORE {}] - REINDEX ACTION - AclChangeSetId {} was not found in database, " + + "it has NOT been reindexed", coreName, aclChangeSetId); } } checkShutdown(); @@ -295,7 +313,6 @@ public class AclTracker extends AbstractTracker } checkShutdown(); } - } protected void purgeAcls() throws IOException, JSONException @@ -376,7 +393,8 @@ public class AclTracker extends AbstractTracker state.setCheckedFirstAclTransactionTime(true); LOGGER.info("[CORE {}] - No acl transactions found - no verification required", coreName); - firstChangeSets = client.getAclChangeSets(null, 0L, null, 2000L, 1); + firstChangeSets = client.getAclChangeSets(null, 0L, + null, INITIAL_MAX_ACL_CHANGE_SET_ID, 1); if (!firstChangeSets.getAclChangeSets().isEmpty()) { AclChangeSet firstChangeSet = firstChangeSets.getAclChangeSets().get(0); @@ -388,19 +406,23 @@ public class AclTracker extends AbstractTracker if (!state.isCheckedFirstAclTransactionTime()) { - firstChangeSets = client.getAclChangeSets(null, 0L, null, 2000L, 1); + firstChangeSets = client.getAclChangeSets(null, 0L, + null, INITIAL_MAX_ACL_CHANGE_SET_ID, 1); if (!firstChangeSets.getAclChangeSets().isEmpty()) { AclChangeSet firstAclChangeSet= firstChangeSets.getAclChangeSets().get(0); long firstAclTxId = firstAclChangeSet.getId(); long firstAclTxCommitTime = firstAclChangeSet.getCommitTimeMs(); - int setSize = this.infoSrv.getAclTxDocsSize(""+firstAclTxId, ""+firstAclTxCommitTime); + int setSize = this.infoSrv.getAclTxDocsSize(Long.toString(firstAclTxId), + Long.toString(firstAclTxCommitTime)); if (setSize == 0) { LOGGER.error("[CORE {}] First acl transaction was not found with the correct timestamp.", coreName); - LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match."); - LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database."); + LOGGER.error("SOLR has successfully connected to your repository " + + "however the SOLR indexes and repository database do not match."); + LOGGER.error("If this is a new or rebuilt database your SOLR indexes " + + "also need to be re-built to match the database."); LOGGER.error("You can also check your SOLR connection details in solrcore.properties."); throw new AlfrescoRuntimeException("Initial acl transaction not found with correct timestamp"); } @@ -421,7 +443,8 @@ public class AclTracker extends AbstractTracker { if (firstChangeSets == null) { - firstChangeSets = client.getAclChangeSets(null, 0L, null, 2000L, 1); + firstChangeSets = client.getAclChangeSets(null, 0L, + null, INITIAL_MAX_ACL_CHANGE_SET_ID, 1); } setLastChangeSetIdAndCommitTimeInTrackerState(firstChangeSets.getAclChangeSets(), state); @@ -432,19 +455,23 @@ public class AclTracker extends AbstractTracker AclChangeSet maxAclTxInIndex = this.infoSrv.getMaxAclChangeSetIdAndCommitTimeInIndex(); if (maxAclTxInIndex.getCommitTimeMs() > maxChangeSetCommitTimeInRepo) { - LOGGER.error("[CORE {}] Last acl transaction was found in index with timestamp later than that of repository.", coreName); + LOGGER.error("[CORE {}] Last acl transaction was found in index with " + + "timestamp later than that of repository.", coreName); LOGGER.error("Max Acl Tx In Index: " + maxAclTxInIndex.getId() + ", In Repo: " + maxChangeSetIdInRepo); LOGGER.error("Max Acl Tx Commit Time In Index: " + maxAclTxInIndex.getCommitTimeMs() + ", In Repo: " + maxChangeSetCommitTimeInRepo); - LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match."); - LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database."); + LOGGER.error("SOLR has successfully connected to your repository" + + " however the SOLR indexes and repository database do not match."); + LOGGER.error("If this is a new or rebuilt database your SOLR indexes " + + "also need to be re-built to match the database."); LOGGER.error("You can also check your SOLR connection details in solrcore.properties."); throw new AlfrescoRuntimeException("Last acl transaction found in index with incorrect timestamp"); } else { state.setCheckedLastAclTransactionTime(true); - LOGGER.info("[CORE {}] - Verified last acl transaction timestamp in index less than or equal to that of repository.", coreName); + LOGGER.info("[CORE {}] - Verified last acl transaction timestamp in index less" + + " than or equal to that of repository.", coreName); } } } @@ -467,7 +494,9 @@ public class AclTracker extends AbstractTracker } } - protected AclChangeSets getSomeAclChangeSets(BoundedDeque changeSetsFound, Long fromCommitTime, long timeStep, int maxResults, long endTime) throws AuthenticationException, IOException, JSONException + protected AclChangeSets getSomeAclChangeSets(BoundedDeque changeSetsFound, Long fromCommitTime, + long timeStep, int maxResults, long endTime) + throws AuthenticationException, IOException, JSONException { long actualTimeStep = timeStep; @@ -477,15 +506,17 @@ public class AclTracker extends AbstractTracker Long startTime = fromCommitTime == null ? Long.valueOf(0L) : fromCommitTime; do { - aclChangeSets = client.getAclChangeSets(startTime, null, startTime + actualTimeStep, null, maxResults); + aclChangeSets = client.getAclChangeSets(startTime, null, + startTime + actualTimeStep, null, maxResults); startTime += actualTimeStep; actualTimeStep *= 2; - if(actualTimeStep > TIME_STEP_32_DAYS_IN_MS) + if(actualTimeStep > MAX_TIME_STEP) { - actualTimeStep = TIME_STEP_32_DAYS_IN_MS; + actualTimeStep = MAX_TIME_STEP; } } - while( ((aclChangeSets.getAclChangeSets().size() == 0) && (startTime < endTime)) || ((aclChangeSets.getAclChangeSets().size() > 0) && alreadyFoundChangeSets(changeSetsFound, aclChangeSets))); + while( ((aclChangeSets.getAclChangeSets().size() == 0) && (startTime < endTime)) || + ((aclChangeSets.getAclChangeSets().size() > 0) && alreadyFoundChangeSets(changeSetsFound, aclChangeSets))); return aclChangeSets; @@ -526,12 +557,13 @@ public class AclTracker extends AbstractTracker trackerStats.addAclTime(time); } - public IndexHealthReport checkIndex(Long toTx, Long toAclTx, Long fromTime, Long toTime) + public IndexHealthReport checkIndex(Long toAclTx, Long fromTime, Long toTime) throws AuthenticationException, IOException, JSONException { // DB ACL TX Count long firstChangeSetCommitTimex = 0; - AclChangeSets firstChangeSets = client.getAclChangeSets(null, 0L, null, 2000L, 1); + AclChangeSets firstChangeSets = client.getAclChangeSets(null, 0L, + null, INITIAL_MAX_ACL_CHANGE_SET_ID, 1); if(firstChangeSets.getAclChangeSets().size() > 0) { AclChangeSet firstChangeSet = firstChangeSets.getAclChangeSets().get(0); @@ -549,11 +581,11 @@ public class AclTracker extends AbstractTracker Long minAclTxId = null; long endTime = System.currentTimeMillis() + infoSrv.getHoleRetention(); AclChangeSets aclTransactions; - BoundedDeque changeSetsFound = new BoundedDeque<>(100); + BoundedDeque changeSetsFound = new BoundedDeque<>(ACL_CHANGE_SETS_FOUND_QUEUE_SIZE); DO: do { - aclTransactions = getSomeAclChangeSets(changeSetsFound, lastAclTxCommitTime, TIME_STEP_1_HR_IN_MS, 2000, - endTime); + aclTransactions = getSomeAclChangeSets(changeSetsFound, + lastAclTxCommitTime, timeStep, maxNumberOfAclChangeSets, endTime); for (AclChangeSet set : aclTransactions.getAclChangeSets()) { // include @@ -597,7 +629,8 @@ public class AclTracker extends AbstractTracker try { ArrayList answer = new ArrayList<>(); - AclChangeSets changeSet = client.getAclChangeSets(null, acltxid, null, acltxid+1, 1); + AclChangeSets changeSet = client.getAclChangeSets(null, acltxid, null, + acltxid+1, 1); List acls = client.getAcls(changeSet.getAclChangeSets(), null, Integer.MAX_VALUE); for (Acl acl : acls) { @@ -646,10 +679,9 @@ public class AclTracker extends AbstractTracker { long startElapsed = System.nanoTime(); - - boolean upToDate = false; + AclChangeSets aclChangeSets; - BoundedDeque changeSetsFound = new BoundedDeque(100); + BoundedDeque changeSetsFound = new BoundedDeque<>(ACL_CHANGE_SETS_FOUND_QUEUE_SIZE); long totalAclCount = 0; int aclCount; @@ -672,7 +704,7 @@ public class AclTracker extends AbstractTracker Long fromCommitTime = getChangeSetFromCommitTime(changeSetsFound, state.getLastChangeSetCommitTimeOnServer() == 0 ? state.getLastGoodChangeSetCommitTimeInIndex() : state.getLastChangeSetCommitTimeOnServer()); - aclChangeSets = getSomeAclChangeSets(changeSetsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, 2000, + aclChangeSets = getSomeAclChangeSets(changeSetsFound, fromCommitTime, timeStep, maxNumberOfAclChangeSets, state.getTimeToStopIndexing()); if (aclChangeSets.getAclChangeSets().size() > 0) @@ -690,53 +722,28 @@ public class AclTracker extends AbstractTracker LOGGER.info("{}-[CORE {}] No ACL change set found after lastTxCommitTime {}", Thread.currentThread().getId(), coreName, fromCommitTime); } - - // Ignore indexed ACL Change Sets - aclChangeSets = new AclChangeSets(aclChangeSets.getAclChangeSets().stream() - .filter(changeSet -> { - try - { - boolean isInIndex = (changeSet.getCommitTimeMs() <= state.getLastIndexedChangeSetCommitTime() && - infoSrv.aclChangeSetInIndex(changeSet.getId(), true)); - if (LOGGER.isTraceEnabled()) - { - LOGGER.trace("{}-[CORE {}] Skipping change Set Id {} as it was already indexed", - Thread.currentThread().getId(), coreName, changeSet.getId()); - } - return !isInIndex; - } - catch (IOException e) - { - LOGGER.warn( - "{}-[CORE {}] Error catched while checking if ACL Change Set {} was in index", - Thread.currentThread().getId(), coreName, changeSet.getId(), e); - return true; - } - }) - .collect(Collectors.toList())); - - // Make sure we do not go ahead of where we started - we will check the holes here // correctly next time - if (aclChangeSets.getAclChangeSets().stream().anyMatch(changeSet -> changeSet.getCommitTimeMs() > state.getTimeToStopIndexing())) + if (aclChangeSets.getAclChangeSets() + .stream() + .anyMatch(changeSet -> changeSet.getCommitTimeMs() > state.getTimeToStopIndexing())) { break; } final AtomicInteger counter = new AtomicInteger(); Collection> changeSetBatches = aclChangeSets.getAclChangeSets().stream() - .collect(Collectors.groupingBy(it -> counter.getAndAdd(it.getAclCount()) / changeSetAclsBatchSize)).values(); + .peek(changeSetsFound::add) + .filter(this::isAclChangeSetAlreadyIndexed) + .collect(Collectors.groupingBy(it -> counter.getAndAdd(it.getAclCount()) / changeSetAclsBatchSize)) + .values(); for (List changeSetBatch : changeSetBatches) { aclCount = indexBatchOfChangeSets(changeSetBatch); - for (AclChangeSet indexed : changeSetBatch) - { - changeSetsFound.add(indexed); - } // Update last committed transactions setLastChangeSetIdAndCommitTimeInTrackerState(changeSetBatch, state); indexAclChangeSetAfterWorker(changeSetBatch, state); @@ -757,28 +764,47 @@ public class AclTracker extends AbstractTracker } } - while ((aclChangeSets.getAclChangeSets().size() > 0) && (upToDate == false)); + while ((aclChangeSets.getAclChangeSets().size() > 0)); LOGGER.info("{}-[CORE {}] Tracked {} ACLs", Thread.currentThread().getId(), coreName, totalAclCount); } + private boolean isAclChangeSetAlreadyIndexed(AclChangeSet changeSet) + { + + try + { + boolean isInIndex = (changeSet.getCommitTimeMs() <= state.getLastIndexedChangeSetCommitTime() && + infoSrv.aclChangeSetInIndex(changeSet.getId(), true)); + if (LOGGER.isTraceEnabled()) + { + LOGGER.trace("{}-[CORE {}] Skipping change Set Id {} as it was already indexed", + Thread.currentThread().getId(), coreName, changeSet.getId()); + } + return !isInIndex; + } + catch (IOException e) + { + LOGGER.warn( + "{}-[CORE {}] Error catched while checking if ACL Change Set {} was in index", + Thread.currentThread().getId(), coreName, changeSet.getId(), e); + return true; + } + + } + private void setLastChangeSetIdAndCommitTimeInTrackerState(List aclChangeSets, TrackerState state) { if (!aclChangeSets.isEmpty()) { - Long maxChangeSetCommitTime = aclChangeSets.stream().max(Comparator.comparing(AclChangeSet::getCommitTimeMs)).get().getCommitTimeMs(); - if(maxChangeSetCommitTime != null) - { - state.setLastChangeSetCommitTimeOnServer(maxChangeSetCommitTime); - } - - Long maxChangeSetId = aclChangeSets.stream().max(Comparator.comparing(AclChangeSet::getId)).get().getId(); - if(maxChangeSetId != null) - { - state.setLastChangeSetIdOnServer(maxChangeSetId); - } + long maxChangeSetCommitTime = + aclChangeSets.stream().max(Comparator.comparing(AclChangeSet::getCommitTimeMs)).get().getCommitTimeMs(); + state.setLastChangeSetCommitTimeOnServer(maxChangeSetCommitTime); + + long maxChangeSetId = aclChangeSets.stream().max(Comparator.comparing(AclChangeSet::getId)).get().getId(); + state.setLastChangeSetIdOnServer(maxChangeSetId); } } @@ -804,15 +830,6 @@ public class AclTracker extends AbstractTracker } } - private int getAclCount(List changeSetBatch) - { - int count = 0; - for (AclChangeSet set : changeSetBatch) - { - count += set.getAclCount(); - } - return count; - } /** * Index ACLs from ACL Change Sets contained in changeSetBatch @@ -822,7 +839,8 @@ public class AclTracker extends AbstractTracker * @param changeSetBatch List of ACL Change Sets to be indexed * @return List of ACL Change Set indexed and Count of ACL indexed */ - private int indexBatchOfChangeSets(List changeSetBatch) throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException { + private int indexBatchOfChangeSets(List changeSetBatch) + throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException { // Exclude ACL Change Set with no ACLs inside List nonEmptyChangeSets = changeSetBatch.stream() .filter(set -> set.getAclCount() > 0) diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java index 7506cb2c7..96751218b 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java @@ -92,10 +92,10 @@ public class CascadeTracker extends AbstractTracker implements Tracker { super(p, client, coreName, informationServer, Tracker.Type.CASCADE); - cascadeTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.cascadeTrackerMaxParallelism", + cascadeTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.cascade.tracker.maxParallelism", String.valueOf(DEFAULT_CASCADE_TRACKER_MAX_PARALLELISM))); - cascadeBatchSize = Integer.parseInt(p.getProperty("alfresco.cascadeNodeBatchSize", + cascadeBatchSize = Integer.parseInt(p.getProperty("alfresco.cascade.tracker.nodeBatchSize", String.valueOf(DEFAULT_CASCADE_NODE_BATCH_SIZE)));; forkJoinPool = new ForkJoinPool(cascadeTrackerParallelism); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java index ff2c66fd1..b260fa031 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java @@ -83,7 +83,7 @@ public class ContentTracker extends AbstractTracker implements Tracker contentUpdateBatchSize = Integer.parseInt(p.getProperty("alfresco.contentUpdateBatchSize", String.valueOf(DEFAULT_CONTENT_UPDATE_BATCH_SIZE))); - contentTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.contentTrackerMaxParallelism", + contentTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.content.tracker.maxParallelism", String.valueOf(DEFAULT_CONTENT_TRACKER_MAX_PARALLELISM))); forkJoinPool = new ForkJoinPool(contentTrackerParallelism); 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 439b1c981..1067fdfe2 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 @@ -49,7 +49,8 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -59,6 +60,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.AtomicInteger; import java.util.stream.Collectors; import static org.alfresco.repo.index.shard.ShardMethodEnum.DB_ID_RANGE; @@ -71,26 +73,31 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker { protected final static Logger LOGGER = LoggerFactory.getLogger(MetadataTracker.class); + private static final int METADATA_TRANSACTIONS_FOUND_QUEUE_SIZE = 100; private static final int DEFAULT_METADATA_TRACKER_MAX_PARALLELISM = 32; private static final int DEFAULT_TRANSACTION_DOCS_BATCH_SIZE = 2000; private static final int DEFAULT_MAX_NUMBER_OF_TRANSACTIONS = 2000; private static final int DEFAULT_NODE_BATCH_SIZE = 50; private static final String DEFAULT_INITIAL_TRANSACTION_RANGE = "0-2000"; + private static final long DEFAULT_METADATA_TRACKER_TIMESTEP = TIME_STEP_1_HR_IN_MS; + private static final long INITIAL_MAX_TXN_ID = 2000L; private int matadataTrackerParallelism; private int transactionDocsBatchSize; private int nodeBatchSize; private int maxNumberOfTransactions; + private long timeStep; - private ConcurrentLinkedQueue transactionsToReindex = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue transactionsToIndex = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue transactionsToPurge = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue nodesToReindex = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue nodesToIndex = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue nodesToPurge = new ConcurrentLinkedQueue<>(); - private ConcurrentLinkedQueue queriesToReindex = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue transactionsToReindex = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue transactionsToIndex = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue transactionsToPurge = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue nodesToReindex = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue nodesToIndex = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue nodesToPurge = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue queriesToReindex = new ConcurrentLinkedQueue<>(); - private final boolean isRunningInProduction = !Boolean.parseBoolean(System.getProperty("alfresco.test", "false")); + private final boolean isRunningInProduction = + !Boolean.parseBoolean(System.getProperty("alfresco.test", "false")); private ForkJoinPool forkJoinPool; @@ -160,11 +167,18 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker String.valueOf(DEFAULT_TRANSACTION_DOCS_BATCH_SIZE))); nodeBatchSize = Integer.parseInt(p.getProperty("alfresco.nodeBatchSize", String.valueOf(DEFAULT_NODE_BATCH_SIZE))); - maxNumberOfTransactions = Integer.parseInt(p.getProperty("alfresco.maxNumberOfTransactions", String.valueOf(DEFAULT_MAX_NUMBER_OF_TRANSACTIONS))); - matadataTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.metadataTrackerMaxParallelism", + maxNumberOfTransactions = Integer.parseInt(p.getProperty("alfresco.metadata.tracker.maxNumberOfTransactions", + String.valueOf(DEFAULT_MAX_NUMBER_OF_TRANSACTIONS))); + matadataTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.metadata.tracker.maxParallelism", String.valueOf(DEFAULT_METADATA_TRACKER_MAX_PARALLELISM))); - String[] minTxninitialRangeString = p.getProperty("solr.initial.transaction.range", DEFAULT_INITIAL_TRANSACTION_RANGE).split("-"); + timeStep = Long.parseLong(p.getProperty("alfresco.metadata.tracker.timestep", + String.valueOf(DEFAULT_METADATA_TRACKER_TIMESTEP))); + + String[] minTxninitialRangeString = + p.getProperty("solr.initial.transaction.range", DEFAULT_INITIAL_TRANSACTION_RANGE) + .split("-"); + cascadeTrackerEnabled = informationServer.cascadeTrackingEnabled(); minTxnIdRange = new Pair<>(Long.valueOf(minTxninitialRangeString[0]), Long.valueOf(minTxninitialRangeString[1])); forkJoinPool = new ForkJoinPool(matadataTrackerParallelism); @@ -179,12 +193,13 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker // Try invoking getNextTxCommitTime service try { - client.getNextTxCommitTime(coreName, 0l); + client.getNextTxCommitTime(coreName, 0L); nextTxCommitTimeServiceAvailable = true; } catch (NoSuchMethodException e) { - LOGGER.warn("nextTxCommitTimeService is not available. Upgrade your ACS Repository version in order to use this feature: {} ", e.getMessage()); + LOGGER.warn("nextTxCommitTimeService is not available. " + + "Upgrade your ACS Repository version in order to use this feature: {} ", e.getMessage()); } catch (Exception e) { @@ -196,12 +211,13 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker { try { - client.getTxIntervalCommitTime(coreName, 0l, 0l); + client.getTxIntervalCommitTime(coreName, 0L, 0L); txIntervalCommitTimeServiceAvailable = true; } catch (NoSuchMethodException e) { - LOGGER.warn("txIntervalCommitTimeServiceAvailable is not available. Upgrade your ACS Repository version " + + LOGGER.warn("txIntervalCommitTimeServiceAvailable is not available. " + + "Upgrade your ACS Repository version " + "to use this feature with DB_ID_RANGE sharding: {} ", e.getMessage()); } catch (Exception e) @@ -219,8 +235,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } @Override - protected void doTrack(String iterationId) throws AuthenticationException, IOException, JSONException, EncoderException - { + protected void doTrack(String iterationId) + throws AuthenticationException, IOException, JSONException { // MetadataTracker must wait until ModelTracker has run ModelTracker modelTracker = this.infoSrv.getAdminHandler().getTrackerRegistry().getModelTracker(); if (modelTracker != null && modelTracker.hasModels()) @@ -255,7 +271,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker queriesToReindex.size() > 0; } - private void trackRepository() throws IOException, AuthenticationException, JSONException, EncoderException + private void trackRepository() throws IOException, AuthenticationException, JSONException { checkShutdown(); @@ -303,7 +319,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker state.setCheckedFirstTransactionTime(true); LOGGER.info("No transactions found - no verification required"); - firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), null, minTxnIdRange.getSecond(), 1); + firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), + null, minTxnIdRange.getSecond(), 1); if (!firstTransactions.getTransactions().isEmpty()) { Transaction firstTransaction = firstTransactions.getTransactions().get(0); @@ -321,7 +338,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker // are skipped if they are not related with the range of the Shard. // Getting the minCommitTime for the Shard is enough in order to check // that the first transaction is present. - long minCommitTime = 0l; + long minCommitTime = 0L; if (docRouter instanceof DBIDRangeRouter && txIntervalCommitTimeServiceAvailable) { try @@ -333,28 +350,34 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } catch (NoSuchMethodException e) { - LOGGER.warn("txIntervalCommitTimeServiceAvailable is not available. If you are using DB_ID_RANGE shard method, " - + "upgrade your ACS Repository version in order to use the skip transactions feature: {} ", e.getMessage()); + LOGGER.warn("txIntervalCommitTimeServiceAvailable is not available." + + " If you are using DB_ID_RANGE shard method, " + + "upgrade your ACS Repository version in order to use the skip transactions feature: {} ", + e.getMessage()); } } // When a Shard with DB_ID_RANGE method is empty, minCommitTime is -1. // No firstTransaction checking is required for this case. - if (minCommitTime != -1l) { + if (minCommitTime != -1L) { - firstTransactions = client.getTransactions(minCommitTime, 0L, null, 2000l, 1); + firstTransactions = client.getTransactions(minCommitTime, 0L, + null, INITIAL_MAX_TXN_ID, 1); if (!firstTransactions.getTransactions().isEmpty()) { Transaction firstTransaction = firstTransactions.getTransactions().get(0); long firstTxId = firstTransaction.getId(); long firstTransactionCommitTime = firstTransaction.getCommitTimeMs(); - int setSize = this.infoSrv.getTxDocsSize(""+firstTxId, ""+firstTransactionCommitTime); + int setSize = this.infoSrv.getTxDocsSize(Long.toString(firstTxId), + Long.toString(firstTransactionCommitTime)); if (setSize == 0) { LOGGER.error("First transaction was not found with the correct timestamp."); - LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match."); - LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database."); + LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes" + + " and repository database do not match."); + LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to be " + + "re-built to match the database."); LOGGER.error("You can also check your SOLR connection details in solrcore.properties."); throw new AlfrescoRuntimeException("Initial transaction not found with correct timestamp"); } @@ -376,7 +399,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker { if (firstTransactions == null) { - firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), null, minTxnIdRange.getSecond(), 1); + firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), + null, minTxnIdRange.getSecond(), 1); } setLastTxCommitTimeAndTxIdInTrackerState(firstTransactions); @@ -391,8 +415,10 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker LOGGER.error("Max Tx In Index: " + maxTxInIndex.getId() + ", In Repo: " + maxTxnIdInRepo); LOGGER.error("Max Tx Commit Time In Index: " + maxTxInIndex.getCommitTimeMs() + ", In Repo: " + maxTxnCommitTimeInRepo); - LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match."); - LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database."); + LOGGER.error("SOLR has successfully connected to your repository however the SOLR indexes" + + " and repository database do not match."); + LOGGER.error("If this is a new or rebuilt database your SOLR indexes also need to " + + "be re-built to match the database."); LOGGER.error("You can also check your SOLR connection details in solrcore.properties."); throw new AlfrescoRuntimeException("Last transaction found in index with incorrect timestamp"); } @@ -416,20 +442,22 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker Long transactionId = transactionsToIndex.poll(); if (transactionId != null) { - Transactions transactions = client.getTransactions(null, transactionId, null, transactionId+1, 1); - if ((transactions.getTransactions().size() > 0) && (transactionId.equals(transactions.getTransactions().get(0).getId()))) + Transactions transactions = client.getTransactions(null, transactionId, + null, transactionId + 1, 1); + if ((transactions.getTransactions().size() > 0) && + (transactionId.equals(transactions.getTransactions().get(0).getId()))) { Transaction info = transactions.getTransactions().get(0); GetNodesParameters gnp = new GetNodesParameters(); - ArrayList txs = new ArrayList(); + ArrayList txs = new ArrayList<>(); txs.add(info.getId()); gnp.setTransactionIds(txs); gnp.setStoreProtocol(storeRef.getProtocol()); gnp.setStoreIdentifier(storeRef.getIdentifier()); updateShardProperty(); - shardProperty.ifPresent(p -> gnp.setShardProperty(p)); + shardProperty.ifPresent(gnp::setShardProperty); gnp.setCoreName(coreName); @@ -455,27 +483,14 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } else { - LOGGER.info("INDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", transactionId); - } - } - - if (docCount > batchCount) - { - if(this.infoSrv.getRegisteredSearcherCount() < getMaxLiveSearchers()) - { - checkShutdown(); - long endElapsed = System.nanoTime(); - trackerStats.addElapsedNodeTime(docCount, endElapsed-startElapsed); - startElapsed = endElapsed; - docCount = 0; - requiresCommit = false; + LOGGER.info("INDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", + transactionId); } } } - if (requiresCommit || (docCount > 0)) + if (requiresCommit) { checkShutdown(); - //this.infoSrv.commit(); long endElapsed = System.nanoTime(); trackerStats.addElapsedNodeTime(docCount, endElapsed - startElapsed); } @@ -520,13 +535,15 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker // make sure it is cleaned out so we do not miss deletes this.infoSrv.deleteByTransactionId(transactionId); - Transactions transactions = client.getTransactions(null, transactionId, null, transactionId+1, 1); - if ((transactions.getTransactions().size() > 0) && (transactionId.equals(transactions.getTransactions().get(0).getId()))) + Transactions transactions = client.getTransactions(null, transactionId, + null, transactionId+1, 1); + if ((transactions.getTransactions().size() > 0) && + (transactionId.equals(transactions.getTransactions().get(0).getId()))) { Transaction info = transactions.getTransactions().get(0); this.infoSrv.dirtyTransaction(info.getId()); GetNodesParameters gnp = new GetNodesParameters(); - ArrayList txs = new ArrayList(); + ArrayList txs = new ArrayList<>(); txs.add(info.getId()); gnp.setTransactionIds(txs); gnp.setStoreProtocol(storeRef.getProtocol()); @@ -550,7 +567,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } else { - LOGGER.info("REINDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", transactionId); + LOGGER.info("REINDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", + transactionId); } } @@ -629,7 +647,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } - private void purgeTransactions() throws IOException, AuthenticationException, JSONException + private void purgeTransactions() throws IOException, JSONException { boolean requiresCommit = false; while (transactionsToPurge.peek() != null) @@ -651,7 +669,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } } - private void purgeNodes() throws IOException, AuthenticationException, JSONException + private void purgeNodes() throws IOException, JSONException { while (nodesToPurge.peek() != null) { @@ -715,22 +733,21 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } protected Transactions getSomeTransactions(BoundedDeque txnsFound, Long fromCommitTime, long timeStep, - int maxResults, long endTime) throws AuthenticationException, IOException, JSONException, EncoderException, NoSuchMethodException + int maxResults, long endTime) + throws AuthenticationException, IOException, JSONException, EncoderException, NoSuchMethodException { - long actualTimeStep = timeStep; - ShardState shardstate = getShardState(); Transactions transactions; // step forward in time until we find something or hit the time bound // max id unbounded - Long startTime = fromCommitTime == null ? 0L : fromCommitTime; + long startTime = fromCommitTime == null ? 0L : fromCommitTime; if(startTime == 0) { return client.getTransactions(startTime, null, - startTime + actualTimeStep, + startTime + timeStep, null, maxResults, shardstate); @@ -738,8 +755,9 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker do { - transactions = client.getTransactions(startTime, null, startTime + actualTimeStep, null, maxResults, shardstate); - startTime += actualTimeStep; + transactions = client.getTransactions(startTime, null, startTime + timeStep, + null, maxResults, shardstate); + startTime += timeStep; // If no transactions are found, advance the time window to the next available transaction commit time if (nextTxCommitTimeServiceAvailable && transactions.getTransactions().size() == 0) @@ -749,7 +767,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker { LOGGER.info("{}-[CORE {}] Advancing transactions from {} to {}", Thread.currentThread().getId(), coreName, startTime, nextTxCommitTime); - transactions = client.getTransactions(nextTxCommitTime, null, nextTxCommitTime + actualTimeStep, null, maxResults, shardstate); + transactions = client.getTransactions(nextTxCommitTime, null, + nextTxCommitTime + timeStep, null, maxResults, shardstate); } } @@ -807,7 +826,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker fromCommitTime = shardMinCommitTime; } - Transactions transactions = getSomeTransactions(txnsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions, + Transactions transactions = getSomeTransactions(txnsFound, fromCommitTime, timeStep, maxNumberOfTransactions, state.getTimeToStopIndexing()); @@ -819,7 +838,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker latestTransaction.setCommitTimeMs(transactions.getMaxTxnCommitTime()); latestTransaction.setId(transactions.getMaxTxnId()); transactions = new Transactions( - Arrays.asList(latestTransaction), + Collections.singletonList(latestTransaction), transactions.getMaxTxnCommitTime(), transactions.getMaxTxnId()); } @@ -827,81 +846,49 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker return transactions; } - /** - * Remove transactions already present in SOLR index - * - * @param transactions List of transactions to be indexed - * @return List of transactions not indexed in SOLR index - */ - private Transactions removeIndexedTransactions(Transactions transactions) + private boolean isTransactionIndexed(Transaction transaction) { - return new Transactions(transactions.getTransactions().stream() - .filter(transaction -> { - try - { - boolean isInIndex = (transaction.getCommitTimeMs() <= state.getLastIndexedTxCommitTime() && - infoSrv.txnInIndex(transaction.getId(), true)); - if (LOGGER.isTraceEnabled()) - { - LOGGER.trace("{}-[CORE {}] Skipping Transaction Id {} as it was already indexed", - Thread.currentThread().getId(), coreName, transaction.getId()); - } - return !isInIndex; - } - catch (IOException e) - { - LOGGER.warn( - "{}-[CORE {}] Error catched while checking if Transaction Id {} was in index", - Thread.currentThread().getId(), coreName, transaction.getId(), e); - return true; - } - }) - .collect(Collectors.toList())); - + try + { + boolean isInIndex = (transaction.getCommitTimeMs() <= state.getLastIndexedTxCommitTime() && + infoSrv.txnInIndex(transaction.getId(), true)); + if (LOGGER.isTraceEnabled()) + { + LOGGER.trace("{}-[CORE {}] Skipping Transaction Id {} as it was already indexed", + Thread.currentThread().getId(), coreName, transaction.getId()); + } + return !isInIndex; + } + catch (IOException e) + { + LOGGER.warn( + "{}-[CORE {}] Error catched while checking if Transaction Id {} was in index", + Thread.currentThread().getId(), coreName, transaction.getId(), e); + return true; + } } - //fixme remove - /** - * Keep only transactions previous to node transaction Id - * - * @param transactions List of transactions from Repository - * @param node Last Node indexed in the cycle - * @return Filtered list of transactions - */ - private Transactions filterTransactionsByNode(Transactions transactions, Node node) - { - return new Transactions(transactions.getTransactions().stream() - .filter(transaction -> { - return transaction.getId() < node.getTxnId(); - }) - .collect(Collectors.toList())); - } - /** * Indexing new transactions from repository in batches of "transactionDocsBatchSize" size. * * Additionally, the nodes inside a transaction batch are indexed in batches of "nodeBatchSize" size. * - * @throws AuthenticationException * @throws IOException * @throws JSONException - * @throws EncoderException */ - protected void trackTransactions() throws AuthenticationException, IOException, JSONException, EncoderException + protected void trackTransactions() throws IOException, JSONException { long startElapsed = System.nanoTime(); - boolean upToDate = false; Transactions transactions; - BoundedDeque txnsFound = new BoundedDeque(100); + BoundedDeque txnsFound = new BoundedDeque<>(METADATA_TRANSACTIONS_FOUND_QUEUE_SIZE); int totalUpdatedDocs = 0; LOGGER.info("{}-[CORE {}] Starting metadata tracker execution", Thread.currentThread().getId(), coreName); do { - try { /* @@ -918,7 +905,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker this.state = getTrackerState(); Long fromCommitTime = getTxFromCommitTime(txnsFound, - state.getLastIndexedTxCommitTime() == 0 ? state.getLastGoodTxCommitTimeInIndex() : state.getLastIndexedTxCommitTime()); + state.getLastIndexedTxCommitTime() == 0 ? state.getLastGoodTxCommitTimeInIndex() + : state.getLastIndexedTxCommitTime()); // Get transaction list to be indexed if (docRouter instanceof DBIDRangeRouter && txIntervalCommitTimeServiceAvailable) @@ -927,12 +915,10 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } else { - transactions = getSomeTransactions(txnsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions, + transactions = getSomeTransactions(txnsFound, fromCommitTime, timeStep, maxNumberOfTransactions, state.getTimeToStopIndexing()); } - // Remove transactions already indexed - transactions = removeIndexedTransactions(transactions); if (transactions.getTransactions().size() > 0) { @@ -949,48 +935,45 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker LOGGER.info("{}-[CORE {}] No transaction found after lastTxCommitTime {}", Thread.currentThread().getId(), coreName, - ((txnsFound.size() > 0) ? txnsFound.getLast().getCommitTimeMs() : state.getLastIndexedTxCommitTime())); + ((txnsFound.size() > 0) ? txnsFound.getLast().getCommitTimeMs() + : state.getLastIndexedTxCommitTime())); } - // Group the transactions in batches of transactionDocsBatchSize (or less) - List> txBatches = new ArrayList<>(); - List txBatch = new ArrayList<>(); - for (Transaction info : transactions.getTransactions()) { - - if (LOGGER.isTraceEnabled()) - { - LOGGER.trace("{}-[CORE {}] Tracking {} Transactions. Current Transaction Id to be indexed: {}", - Thread.currentThread().getId(), coreName, transactions.getTransactions().size(), info.getId()); - } - - // Make sure we do not go ahead of where we started - we will check the holes here - // correctly next time - if (info.getCommitTimeMs() > state.getTimeToStopIndexing()) { - upToDate = true; - break; - } - - txBatch.add(info); - if (getUpdateAndDeleteCount(txBatch) > transactionDocsBatchSize) { - txBatches.add(txBatch); - txBatch = new ArrayList<>(); - } - } - if (!txBatch.isEmpty()) + // 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())) { - txBatches.add(txBatch); + break; } + long transaction_number = transactions.getTransactions().size(); + final AtomicInteger counter = new AtomicInteger(); + Collection> txBatches = transactions.getTransactions().stream() + .peek(txnsFound::add) + .filter(this::isTransactionIndexed) + .peek(transaction -> { + if (LOGGER.isTraceEnabled()) + { + LOGGER.trace("{}-[CORE {}] Tracking {} Transactions. Current Transaction Id to be indexed: {}", + Thread.currentThread().getId(), coreName, transaction_number, transaction.getId()); + } + }) + .collect(Collectors.groupingBy(transaction -> counter.getAndAdd( + (int) (transaction.getDeletes() + transaction.getUpdates())) / transactionDocsBatchSize)) + .values(); + // Index batches of transactions and the nodes updated or deleted within the transaction for (List batch : txBatches) { // Index nodes contained in the transactions - int docCount = indexBatchOfTransactions(batch, totalUpdatedDocs); + int docCount = indexBatchOfTransactions(batch); totalUpdatedDocs += docCount; // Add the transactions as found to avoid processing them again in the next iteration - batch.forEach(transaction -> txnsFound.add(transaction)); + batch.forEach(txnsFound::add); // Index the transactions indexTransactionsAfterWorker(batch); @@ -999,10 +982,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker startElapsed = endElapsed; } - - setLastTxCommitTimeAndTxIdInTrackerState(transactions); - } catch(Exception e) { @@ -1014,7 +994,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } } - while ((transactions.getTransactions().size() > 0) && (upToDate == false)); + while ((transactions.getTransactions().size() > 0)); LOGGER.info("{}-[CORE {}] Tracked {} DOCs", Thread.currentThread().getId(), coreName, totalUpdatedDocs); } @@ -1062,20 +1042,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker txsIndexed.clear(); } - /** - * Return the number of updated and deleted nodes in a list of transactions - * @param txs List of transactions - * @return Number of updated and deleted nodes - */ - private long getUpdateAndDeleteCount(List txs) - { - long count = 0; - for (Transaction tx : txs) - { - count += (tx.getUpdates() + tx.getDeletes()); - } - return count; - } /** * Index a batch of transactions. @@ -1084,23 +1050,20 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker * the metadata of the nodes in smaller invocations to Repository * * @param txBatch Batch of transactions to be indexed - * @param indexedNodes Number of nodes indexed in this Tracker execution - * * @return Number of nodes indexed and last node indexed * * @throws AuthenticationException * @throws IOException * @throws JSONException */ - private int indexBatchOfTransactions(List txBatch, int indexedNodes) throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException { + private int indexBatchOfTransactions(List txBatch) + throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException { // Skip transactions without modifications (updates, deletes) - ArrayList nonEmptyTxs = new ArrayList<>(txBatch.size()); ArrayList txIds = new ArrayList<>(); for (Transaction tx : txBatch) { if (tx.getUpdates() > 0 || tx.getDeletes() > 0) { - nonEmptyTxs.add(tx); txIds.add(tx.getId()); } } @@ -1118,22 +1081,18 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker if (LOGGER.isDebugEnabled()) { - LOGGER.debug("{}-[CORE {}] Found {} Nodes to be indexed from Transactions: {}", Thread.currentThread().getId(), - coreName, nodes.size(), txIds); + LOGGER.debug("{}-[CORE {}] Found {} Nodes to be indexed from Transactions: {}", + Thread.currentThread().getId(), coreName, nodes.size(), txIds); } // Group the nodes in batches of nodeBatchSize (or less) List> nodeBatches = Lists.partition(nodes, nodeBatchSize); - Integer processedNodes = forkJoinPool.submit(() -> + return forkJoinPool.submit(() -> nodeBatches.parallelStream().map(batch -> { new NodeIndexWorker(batch, infoSrv).run(); return batch.size(); }).reduce(0, Integer::sum)).get(); - - - - return processedNodes; } @@ -1260,7 +1219,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker try { GetNodesParameters gnp = new GetNodesParameters(); - ArrayList txs = new ArrayList(); + ArrayList txs = new ArrayList<>(); txs.add(txid); gnp.setTransactionIds(txs); gnp.setStoreProtocol(storeRef.getProtocol()); @@ -1268,26 +1227,19 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker gnp.setCoreName(coreName); return client.getNodes(gnp, Integer.MAX_VALUE); } - catch (IOException e) - { - throw new AlfrescoRuntimeException("Failed to get nodes", e); - } - catch (JSONException e) - { - throw new AlfrescoRuntimeException("Failed to get nodes", e); - } - catch (AuthenticationException e) + catch (IOException | AuthenticationException | JSONException e) { throw new AlfrescoRuntimeException("Failed to get nodes", e); } } - public IndexHealthReport checkIndex(Long toTx, Long toAclTx, Long fromTime, Long toTime) + public IndexHealthReport checkIndex(Long toTx, Long fromTime, Long toTime) throws IOException, AuthenticationException, JSONException, EncoderException, NoSuchMethodException { // DB TX Count long firstTransactionCommitTime = 0; - Transactions firstTransactions = client.getTransactions(null, 0L, null, 2000l, 1); + Transactions firstTransactions = client.getTransactions(null, 0L, + null, INITIAL_MAX_TXN_ID, 1); if(firstTransactions.getTransactions().size() > 0) { Transaction firstTransaction = firstTransactions.getTransactions().get(0); @@ -1295,7 +1247,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker } IOpenBitSet txIdsInDb = infoSrv.getOpenBitSetInstance(); - Long lastTxCommitTime = Long.valueOf(firstTransactionCommitTime); + long lastTxCommitTime = firstTransactionCommitTime; if (fromTime != null) { lastTxCommitTime = fromTime; @@ -1304,24 +1256,24 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker Long minTxId = null; Transactions transactions; - BoundedDeque txnsFound = new BoundedDeque(100); + BoundedDeque txnsFound = new BoundedDeque<>(METADATA_TRANSACTIONS_FOUND_QUEUE_SIZE); long endTime = System.currentTimeMillis() + infoSrv.getHoleRetention(); DO: do { - transactions = getSomeTransactions(txnsFound, lastTxCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions, endTime); + transactions = getSomeTransactions(txnsFound, lastTxCommitTime, timeStep, maxNumberOfTransactions, endTime); for (Transaction info : transactions.getTransactions()) { // include if (toTime != null) { - if (info.getCommitTimeMs() > toTime.longValue()) + if (info.getCommitTimeMs() > toTime) { break DO; } } if (toTx != null) { - if (info.getId() > toTx.longValue()) + if (info.getId() > toTx) { break DO; } @@ -1382,9 +1334,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker infoSrv.clearProcessedTransactions(); } - /** - * @param query - */ public void addQueryToReindex(String query) { this.queriesToReindex.offer(query); diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/AlfrescoSolrUtils.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/AlfrescoSolrUtils.java index 17d0166eb..c910878ab 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/AlfrescoSolrUtils.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/AlfrescoSolrUtils.java @@ -135,18 +135,17 @@ public class AlfrescoSolrUtils */ public static Transaction getTransaction(int deletes, int updates) { - long txnCommitTime = System.currentTimeMillis(); - Transaction transaction = new Transaction(); - transaction.setCommitTimeMs(txnCommitTime); - transaction.setId(generateId()); - transaction.setDeletes(deletes); - transaction.setUpdates(updates); - return transaction; + return getTransaction(deletes, updates, generateId()); } public static Transaction getTransaction(int deletes, int updates, long id) { - long txnCommitTime = System.currentTimeMillis(); + return getTransaction(deletes, updates, id, System.currentTimeMillis()); + } + + public static Transaction getTransaction(int deletes, int updates, long id, long timestamp) + { + long txnCommitTime = timestamp; Transaction transaction = new Transaction(); transaction.setCommitTimeMs(txnCommitTime); transaction.setId(id); @@ -309,6 +308,7 @@ public class AlfrescoSolrUtils Acl acl = new Acl(aclChangeSet.getId(), aclId); return acl; } + /** * Get an AclChangeSet * @param aclCount @@ -316,14 +316,17 @@ public class AlfrescoSolrUtils */ public static AclChangeSet getAclChangeSet(int aclCount) { - AclChangeSet aclChangeSet = new AclChangeSet(generateId(), System.currentTimeMillis(), aclCount); - return aclChangeSet; + return new AclChangeSet(generateId(), System.currentTimeMillis(), aclCount); } public static AclChangeSet getAclChangeSet(int aclCount, long id) { - AclChangeSet aclChangeSet = new AclChangeSet(id, System.currentTimeMillis(), aclCount); - return aclChangeSet; + return new AclChangeSet(id, System.currentTimeMillis(), aclCount); + } + + public static AclChangeSet getAclChangeSet(int aclCount, long id, long timestamp) + { + return new AclChangeSet(id, timestamp, aclCount); } private static AtomicLong id = new AtomicLong(System.currentTimeMillis()); diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java index bb3b1fa52..24d744f00 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java @@ -29,7 +29,6 @@ package org.alfresco.solr; import static org.alfresco.solr.AlfrescoSolrUtils.createCoreUsingTemplate; import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope; -import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering; import org.alfresco.solr.basics.RandomSupplier; import org.alfresco.solr.client.SOLRAPIQueueClient; import org.apache.commons.io.FileUtils; @@ -98,7 +97,10 @@ import java.util.concurrent.atomic.AtomicInteger; public abstract class SolrITInitializer extends SolrTestCaseJ4 { private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - + protected static final int DEFAULT_CONNECTION_TIMEOUT1 = DEFAULT_CONNECTION_TIMEOUT; + protected static final int CLIENT_SO_TIMEOUT = 90000; + protected final static int INDEX_TIMEOUT = 100000; + private static AtomicInteger nodeCnt; protected static boolean useExplicitNodeNames; @@ -115,8 +117,6 @@ public abstract class SolrITInitializer extends SolrTestCaseJ4 //Standalone Tests protected static SolrCore defaultCore; - protected static final int clientConnectionTimeout = DEFAULT_CONNECTION_TIMEOUT; - protected static final int clientSoTimeout = 90000; protected static final String id = "id"; @@ -501,8 +501,8 @@ public abstract class SolrITInitializer extends SolrTestCaseJ4 try { HttpSolrClient client = new HttpSolrClient(url); - client.setConnectionTimeout(clientConnectionTimeout); - client.setSoTimeout(clientSoTimeout); + client.setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT1); + client.setSoTimeout(CLIENT_SO_TIMEOUT); client.setDefaultMaxConnectionsPerHost(100); client.setMaxTotalConnections(100); return client; diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoSolrTrackerRaceIT.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoSolrTrackerRaceIT.java index 631cd7363..bc2cec9d2 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoSolrTrackerRaceIT.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoSolrTrackerRaceIT.java @@ -66,7 +66,7 @@ public class DistributedAlfrescoSolrTrackerRaceIT extends AbstractAlfrescoDistri @BeforeClass public static void initData() throws Throwable { - initSolrServers(2, "DistributedAlfrescoSolrTrackerRaceTest", null); + initSolrServers(2, "DistributedAlfrescoSolrTrackerRaceIT", null); } @AfterClass @@ -76,25 +76,23 @@ public class DistributedAlfrescoSolrTrackerRaceIT extends AbstractAlfrescoDistri } @Test - public void testTracker() throws Exception - { + public void testTracker() throws Exception { putHandleDefaults(); - AclChangeSet aclChangeSet = getAclChangeSet(1); + AclChangeSet aclChangeSet = getAclChangeSet(2, 1, System.currentTimeMillis() - + AbstractTracker.TIME_STEP_32_DAYS_IN_MS); Acl acl = getAcl(aclChangeSet); Acl acl2 = getAcl(aclChangeSet); - AclReaders aclReaders = getAclReaders(aclChangeSet, acl, singletonList("joel"), singletonList("phil"), null); - AclReaders aclReaders2 = getAclReaders(aclChangeSet, acl2, singletonList("jim"), singletonList("phil"), null); + AclReaders aclReaders = + getAclReaders(aclChangeSet, acl, singletonList("joel"), singletonList("phil"), null); + AclReaders aclReaders2 = + getAclReaders(aclChangeSet, acl2, singletonList("jim"), singletonList("phil"), null); // Transaction between [1-2000] is required, when greater value checking the core will fail - Transaction txn = getTransaction(0, 2, 1); - long txnCommitTimeMs = txn.getCommitTimeMs(); - - // Subtract from the commit time to go beyond hole retention - long backdatedCommitTimeMs = txnCommitTimeMs - 4600000; - txn.setCommitTimeMs(backdatedCommitTimeMs); + Transaction txn = getTransaction(0, 2, 1, + System.currentTimeMillis() - AbstractTracker.TIME_STEP_32_DAYS_IN_MS); //Next create two nodes to update for the transaction Node folderNode = getNode(txn, acl, Node.SolrApiNodeStatus.UPDATED); @@ -103,9 +101,12 @@ public class DistributedAlfrescoSolrTrackerRaceIT extends AbstractAlfrescoDistri // Next, create the node metadata for each node. // Note: the error node metadata will cause an exception. - NodeMetaData folderMetaData = getNodeMetaData(folderNode, txn, acl, "mike", null, false); - NodeMetaData fileMetaData = getNodeMetaData(fileNode, txn, acl, "mike", ancestors(folderMetaData.getNodeRef()), false); - NodeMetaData errorMetaData = getNodeMetaData(errorNode, txn, acl, "lisa", ancestors(folderMetaData.getNodeRef()), true); + NodeMetaData folderMetaData = + getNodeMetaData(folderNode, txn, acl, "mike", null, false); + NodeMetaData fileMetaData = getNodeMetaData(fileNode, txn, acl, "mike", + ancestors(folderMetaData.getNodeRef()), false); + NodeMetaData errorMetaData = getNodeMetaData(errorNode, txn, acl, "lisa", + ancestors(folderMetaData.getNodeRef()), true); // Index the transaction, nodes, and nodeMetaDatas. // Note that the content is automatically created by the test framework. @@ -113,25 +114,34 @@ public class DistributedAlfrescoSolrTrackerRaceIT extends AbstractAlfrescoDistri indexAclChangeSet(aclChangeSet, asList(acl, acl2), asList(aclReaders, aclReaders2)); BooleanQuery.Builder builder = new BooleanQuery.Builder(); - builder.add(new BooleanClause(new TermQuery(new Term(QueryConstants.FIELD_SOLR4_ID, "TRACKER!STATE!ACLTX")), BooleanClause.Occur.MUST)); - builder.add(new BooleanClause(LegacyNumericRangeQuery.newLongRange(QueryConstants.FIELD_S_ACLTXID, aclChangeSet.getId(), aclChangeSet.getId() + 1, true, false), BooleanClause.Occur.MUST)); + builder.add(new BooleanClause(new TermQuery(new Term(QueryConstants.FIELD_SOLR4_ID, "TRACKER!STATE!ACLTX")), + BooleanClause.Occur.MUST)); + + builder.add(new BooleanClause(LegacyNumericRangeQuery.newLongRange(QueryConstants.FIELD_S_ACLTXID, + aclChangeSet.getId(), aclChangeSet.getId() + 1, true, false), + BooleanClause.Occur.MUST)); + BooleanQuery waitForQuery = builder.build(); - waitForDocCountAllCores(waitForQuery, 1, 80000); + waitForDocCountAllCores(waitForQuery, 1, INDEX_TIMEOUT); // This ACL should have one record in each core with DBID sharding - waitForDocCountAllCores(new TermQuery(new Term(QueryConstants.FIELD_READER, "jim")), 1, 80000); + waitForDocCountAllCores(new TermQuery(new Term(QueryConstants.FIELD_READER, "jim")), 1, INDEX_TIMEOUT); // We should have 2 document in totals (1 folder and 1 file) - waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 2, 100000); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + "world")), 2, INDEX_TIMEOUT); // There should be 1 with the folder node identifier. - waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", Long.toString(folderNode.getId()))), 1, 80000); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + Long.toString(folderNode.getId()))), 1, INDEX_TIMEOUT); // There should be 1 with the file node identifier. - waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", Long.toString(fileNode.getId()))), 1, 80000); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + Long.toString(fileNode.getId()))), 1, INDEX_TIMEOUT); // and last but not least, the error node shouldn't be in the index. - waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", Long.toString(errorNode.getId()))), 0, 80000); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + Long.toString(errorNode.getId()))), 0, INDEX_TIMEOUT); // This will run the same query on the control client and the cluster and compare the result. query(getDefaultTestClient(), diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoTrackerWithDelayedTransactionsIT.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoTrackerWithDelayedTransactionsIT.java new file mode 100644 index 000000000..60258626c --- /dev/null +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoTrackerWithDelayedTransactionsIT.java @@ -0,0 +1,169 @@ +/* + * #%L + * Alfresco Search Services + * %% + * Copyright (C) 2005 - 2020 Alfresco Software Limited + * %% + * This file is part of the Alfresco software. + * If the software was purchased under a paid Alfresco license, the terms of + * the paid license agreement will prevail. Otherwise, the software is + * provided under the following open source license terms: + * + * Alfresco is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * Alfresco is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with Alfresco. If not, see . + * #L% + */ + +package org.alfresco.solr.tracker; + +import org.alfresco.repo.search.adaptor.lucene.QueryConstants; +import org.alfresco.solr.AbstractAlfrescoDistributedIT; +import org.alfresco.solr.client.Acl; +import org.alfresco.solr.client.AclChangeSet; +import org.alfresco.solr.client.AclReaders; +import org.alfresco.solr.client.Node; +import org.alfresco.solr.client.NodeMetaData; +import org.alfresco.solr.client.Transaction; +import org.apache.lucene.index.Term; +import org.apache.lucene.search.BooleanClause; +import org.apache.lucene.search.BooleanQuery; +import org.apache.lucene.search.LegacyNumericRangeQuery; +import org.apache.lucene.search.TermQuery; +import org.apache.solr.SolrTestCaseJ4; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.util.List; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static org.alfresco.solr.AlfrescoSolrUtils.ancestors; +import static org.alfresco.solr.AlfrescoSolrUtils.getAcl; +import static org.alfresco.solr.AlfrescoSolrUtils.getAclChangeSet; +import static org.alfresco.solr.AlfrescoSolrUtils.getAclReaders; +import static org.alfresco.solr.AlfrescoSolrUtils.getNode; +import static org.alfresco.solr.AlfrescoSolrUtils.getNodeMetaData; +import static org.alfresco.solr.AlfrescoSolrUtils.getTransaction; +import static org.alfresco.solr.AlfrescoSolrUtils.indexAclChangeSet; + +@SolrTestCaseJ4.SuppressSSL +public class DistributedAlfrescoTrackerWithDelayedTransactionsIT extends AbstractAlfrescoDistributedIT { + + private static NodeMetaData folderMetaData; + private static AclChangeSet aclChangeSet; + + @BeforeClass + public static void initData() throws Throwable + { + initializeDataBeforeServerCreation(); + initSolrServers(2, "DistributedAlfrescoTrackerWithDelayedTransactionsIT", + null); + } + + + @AfterClass + public static void destroyData() + { + dismissSolrServers(); + } + + private static void initializeDataBeforeServerCreation() + { + putHandleDefaults(); + + aclChangeSet = getAclChangeSet(2, 1, + System.currentTimeMillis() - AbstractTracker.TIME_STEP_32_DAYS_IN_MS); + + Acl acl = getAcl(aclChangeSet); + Acl acl2 = getAcl(aclChangeSet); + + AclReaders aclReaders = + getAclReaders(aclChangeSet, acl, singletonList("joel"), singletonList("phil"), null); + AclReaders aclReaders2 = + getAclReaders(aclChangeSet, acl2, singletonList("jim"), singletonList("phil"), null); + + // Transaction between [1-2000] is required, when greater value checking the core will fail + Transaction txn = getTransaction(0, 2, 1, + System.currentTimeMillis() - AbstractTracker.TIME_STEP_32_DAYS_IN_MS); + + //Next create two nodes to update for the transaction + Node folderNode = getNode(txn, acl, Node.SolrApiNodeStatus.UPDATED); + Node fileNode = getNode(txn, acl, Node.SolrApiNodeStatus.UPDATED); + + folderMetaData = getNodeMetaData(folderNode, txn, acl, "mike", null, false); + NodeMetaData fileMetaData = getNodeMetaData(fileNode, txn, acl, "mike", + ancestors(folderMetaData.getNodeRef()), false); + + indexAclChangeSet(aclChangeSet, asList(acl, acl2), asList(aclReaders, aclReaders2)); + indexTransaction(txn, List.of(folderNode, fileNode), List.of(folderMetaData, fileMetaData)); + } + + @Test + public void testTracker() throws Exception { + + BooleanQuery.Builder builder = new BooleanQuery.Builder(); + builder.add(new BooleanClause(new TermQuery(new Term(QueryConstants.FIELD_SOLR4_ID, "TRACKER!STATE!ACLTX")), + BooleanClause.Occur.MUST)); + builder.add(new BooleanClause(LegacyNumericRangeQuery.newLongRange(QueryConstants.FIELD_S_ACLTXID, + aclChangeSet.getId(), aclChangeSet.getId() + 1, true, false), + BooleanClause.Occur.MUST)); + + BooleanQuery waitForQuery = builder.build(); + waitForDocCountAllCores(waitForQuery, 1, INDEX_TIMEOUT); + + // This ACL should have one record in each core with DBID sharding + waitForDocCountAllCores(new TermQuery(new Term(QueryConstants.FIELD_READER, "jim")), + 1, INDEX_TIMEOUT); + + // We should have 2 document in totals (1 folder and 1 file) + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + "world")), 2, INDEX_TIMEOUT); + + + // This will run the same query on the control client and the cluster and compare the result. + query(getDefaultTestClient(), + true, + "{\"locales\":[\"en\"], \"templates\": [{\"name\":\"t1\", \"template\":\"%cm:content\"}]}", + params("q", "t1:world", "qt", "/afts", "shards.qt", "/afts", "start", "0", "rows", "6", "sort", "id asc")); + + + // Index new ACL after 32 days + AclChangeSet lateAclChangeSet = getAclChangeSet(1, 2); + Acl lateAcl = getAcl(lateAclChangeSet); + AclReaders lateAclReaders = + getAclReaders(lateAclChangeSet, lateAcl, singletonList("elia"), singletonList("phil"), null); + indexAclChangeSet(lateAclChangeSet, asList(lateAcl), asList(lateAclReaders)); + + // Index new transaction after 32 days + Transaction lateTransaction = getTransaction(0, 1); + Node lateNode = getNode(lateTransaction, lateAcl, Node.SolrApiNodeStatus.UPDATED); + NodeMetaData lateNodeMetaData = getNodeMetaData(lateNode, lateTransaction, lateAcl, "elia", + ancestors(folderMetaData.getNodeRef()), false); + + // Check new Acl has been indexed + builder = new BooleanQuery.Builder(); + builder.add(new BooleanClause(new TermQuery(new Term(QueryConstants.FIELD_SOLR4_ID, "TRACKER!STATE!ACLTX")), + BooleanClause.Occur.MUST)); + builder.add(new BooleanClause(LegacyNumericRangeQuery.newLongRange(QueryConstants.FIELD_S_ACLTXID, + lateAclChangeSet.getId(), lateAclChangeSet.getId() + 1, true, false), + BooleanClause.Occur.MUST)); + waitForQuery = builder.build(); + waitForDocCountAllCores(waitForQuery, 1, INDEX_TIMEOUT); + + // Check the new node has been indexed + indexTransaction(lateTransaction, List.of(lateNode), List.of(lateNodeMetaData)); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", + "world")), 3, INDEX_TIMEOUT); + } +} diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedCascadeTrackerIT.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedCascadeTrackerIT.java index 980053dfb..ad9ba2474 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedCascadeTrackerIT.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedCascadeTrackerIT.java @@ -77,8 +77,6 @@ public class DistributedCascadeTrackerIT extends AbstractAlfrescoDistributedIT private final String pathChild1 = "pathChild2"; - private final int timeout = 100000; - @Before public void initData() throws Throwable { @@ -139,7 +137,7 @@ public class DistributedCascadeTrackerIT extends AbstractAlfrescoDistributedIT */ indexParentFolderWithCascade(); - waitForDocCount(params("qt", "/afts", "q", "PATH:" + cascadingFirstChild), 1, timeout); + waitForDocCount(params("qt", "/afts", "q", "PATH:" + cascadingFirstChild), 1, INDEX_TIMEOUT); // Check if the path is updated for both the nodes assertShardCount(0, params("qt", "/afts", "q", "PATH:" + cascadingFirstChild), 1); @@ -184,7 +182,7 @@ public class DistributedCascadeTrackerIT extends AbstractAlfrescoDistributedIT /* * Get sure the nodes are indexed correctly in the shards */ - waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 3, timeout); + waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 3, INDEX_TIMEOUT); assertShardCount(0, new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 2); assertShardCount(1, new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 1); }