inteligr8/SearchServices
1
0
Fork 0
mirror of https://github.com/Alfresco/SearchServices.git synced 2026-09-16 18:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

[SEARCH-2278]

Browse Source 
Modified DistributedAlfrescoSolrTrackerRace in order to test an hole of more than an hour between transactions
This commit is contained in:
Elia Porciani
2020-06-03 12:29:08 +01:00
parent 9723fd81b3
commit b530ef731c
11 changed files with 506 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java
+2 -2
View File
@@ -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());
search-services/alfresco-search/src/main/java/org/alfresco/solr/HandlerReportHelper.java
+2 -2
View File
@@ -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<Object> 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());
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java
+122 -104
View File
@@ -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<Long> aclChangeSetsToReindex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclChangeSetsToIndex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclChangeSetsToPurge = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclsToReindex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclsToIndex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclsToPurge = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclChangeSetsToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> aclChangeSetsToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> aclChangeSetsToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> aclsToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> aclsToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> 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<Acl> 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<Acl> acls = client.getAcls(Collections.singletonList(changeSet), null, Integer.MAX_VALUE);
List<Acl> acls = client.getAcls(Collections.singletonList(changeSet),
null, Integer.MAX_VALUE);
for (Acl acl : acls)
{
List<AclReaders> 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<AclChangeSet> changeSetsFound, Long fromCommitTime, long timeStep, int maxResults, long endTime) throws AuthenticationException, IOException, JSONException
protected AclChangeSets getSomeAclChangeSets(BoundedDeque<AclChangeSet> 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<AclChangeSet> changeSetsFound = new BoundedDeque<>(100);
BoundedDeque<AclChangeSet> 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<Long> answer = new ArrayList<>();
AclChangeSets changeSet = client.getAclChangeSets(null, acltxid, null, acltxid+1, 1);
AclChangeSets changeSet = client.getAclChangeSets(null, acltxid, null,
acltxid+1, 1);
List<Acl> 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<AclChangeSet> changeSetsFound = new BoundedDeque<AclChangeSet>(100);
BoundedDeque<AclChangeSet> 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<List<AclChangeSet>> 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<AclChangeSet> 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 {}] <end> 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<AclChangeSet> 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<AclChangeSet> 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<AclChangeSet> changeSetBatch) throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException {
private int indexBatchOfChangeSets(List<AclChangeSet> changeSetBatch)
throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException {
// Exclude ACL Change Set with no ACLs inside
List<AclChangeSet> nonEmptyChangeSets = changeSetBatch.stream()
.filter(set -> set.getAclCount() > 0)
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java
+2 -2
View File
@@ -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);
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java
+1 -1
View File
@@ -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);
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java
+152 -203
View File
@@ -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<Long> transactionsToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> transactionsToReindex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> transactionsToIndex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> transactionsToPurge = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> nodesToReindex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<String> 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<Long> txs = new ArrayList<Long>();
ArrayList<Long> 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<Long> txs = new ArrayList<Long>();
ArrayList<Long> 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<Transaction> 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<Transaction> txnsFound = new BoundedDeque<Transaction>(100);
BoundedDeque<Transaction> 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<List<Transaction>> txBatches = new ArrayList<>();
List<Transaction> 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<List<Transaction>> 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<Transaction> 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<Transaction> 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<Transaction> txBatch, int indexedNodes) throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException {
private int indexBatchOfTransactions(List<Transaction> txBatch)
throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException {
// Skip transactions without modifications (updates, deletes)
ArrayList<Transaction> nonEmptyTxs = new ArrayList<>(txBatch.size());
ArrayList<Long> 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<List<Node>> 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<Long> txs = new ArrayList<Long>();
ArrayList<Long> 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<Transaction> txnsFound = new BoundedDeque<Transaction>(100);
BoundedDeque<Transaction> 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);
search-services/alfresco-search/src/test/java/org/alfresco/solr/AlfrescoSolrUtils.java
+15 -12
View File
@@ -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());
search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java
+6 -6
View File
@@ -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;
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoSolrTrackerRaceIT.java
+33 -23
View File
@@ -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(),
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedAlfrescoTrackerWithDelayedTransactionsIT.java
+169
View File
@@ -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 <http://www.gnu.org/licenses/>.
* #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);
}
}
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedCascadeTrackerIT.java
+2 -4
View File
@@ -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);
}