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-2129: Static semaphore for running and writing Tracker Threads, that avoids parallel execution of the same job.

Browse Source 
Standarization for logger naming and declaration.
This commit is contained in:
Elia Porciani
2020-05-01 08:45:02 +01:00
parent ccdecbdeb2
commit 0875a04a9c
26 changed files with 1349 additions and 1171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
search-services/alfresco-search/src/main/java/org/alfresco/solr/InformationServer.java
+1 -1
View File
@@ -148,7 +148,7 @@ public interface InformationServer extends InformationServerCollectionProvider
IndexHealthReport reportIndexTransactions(Long minTxId, IOpenBitSet txIdsInDb, long maxTxId) throws IOException;
List<TenantAclIdDbId> getDocsWithUncleanContent(int start, int rows) throws IOException;
List<TenantAclIdDbId> getDocsWithUncleanContent() throws IOException;
void updateContentToIndexAndCache(long dbId, String tenant) throws Exception;
search-services/alfresco-search/src/main/java/org/alfresco/solr/SolrInformationServer.java
+6 -2
View File
@@ -663,7 +663,7 @@ public class SolrInformationServer implements InformationServer
}
@Override
public List<TenantAclIdDbId> getDocsWithUncleanContent(int start, int rows) throws IOException
public List<TenantAclIdDbId> getDocsWithUncleanContent() throws IOException
{
RefCounted<SolrIndexSearcher> refCounted = null;
try
@@ -688,7 +688,7 @@ public class SolrInformationServer implements InformationServer
* in current snapshot of the index.
*
* The code below runs every two minutes and purges transactions from the
* cleanContentCache that is more then 20 minutes old.
* cleanContentCache that is more than 20 minutes old.
*
*/
long purgeTime = System.currentTimeMillis();
@@ -721,7 +721,11 @@ public class SolrInformationServer implements InformationServer
DelegatingCollector delegatingCollector = new TxnCacheFilter(cleanContentCache); //Filter transactions that have already been processed.
delegatingCollector.setLastDelegate(collector);
searcher.search(dirtyOrNewContentQuery(), delegatingCollector);
LOGGER.debug("{}-[CORE {}] Processing {} documents with content to be indexed", Thread.currentThread().getId(), core.getName(), collector.getTotalHits());
if(collector.getTotalHits() == 0)
{
return docIds;
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java
+326 -360
View File
@@ -1,361 +1,327 @@
/*
* Copyright (C) 2005-2019 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* 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/>.
*/
package org.alfresco.solr.tracker;
import static java.util.Optional.ofNullable;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.Properties;
import java.util.concurrent.Semaphore;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.solr.IndexTrackingShutdownException;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.TrackerState;
import org.alfresco.solr.client.SOLRAPIClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Abstract base class that provides common {@link Tracker} behaviour.
*
* @author Matt Ward
*/
public abstract class AbstractTracker implements Tracker
{
static final long TIME_STEP_32_DAYS_IN_MS = 1000 * 60 * 60 * 24 * 32L;
static final long TIME_STEP_1_HR_IN_MS = 60 * 60 * 1000L;
static final String SHARD_METHOD_DBID = "DB_ID";
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractTracker.class);
protected Properties props;
protected SOLRAPIClient client;
InformationServer infoSrv;
protected String coreName;
StoreRef storeRef;
long batchCount;
TrackerStats trackerStats;
boolean runPostModelLoadInit = true;
private int maxLiveSearchers;
private volatile boolean shutdown = false;
private Semaphore runLock = new Semaphore(1, true);
private Semaphore writeLock = new Semaphore(1, true);
protected volatile TrackerState state;
protected int shardCount;
protected int shardInstance;
String shardMethod;
protected boolean transformContent;
String shardTemplate;
protected volatile boolean rollback;
/**
* When rollback is set, original error is also gathered in order to provide detailed logging.
*/
protected Throwable rollbackCausedBy;
protected final Type type;
protected final String trackerId;
/*
* A thread handler can be used by subclasses, but they have to intentionally instantiate it.
*/
ThreadHandler threadHandler;
/**
* Default constructor, strictly for testing.
*/
protected AbstractTracker(Type type)
{
this.type = type;
this.trackerId = type + "@" + hashCode();
}
protected AbstractTracker(Properties p, SOLRAPIClient client, String coreName, InformationServer informationServer,Type type)
{
this.props = p;
this.client = client;
this.coreName = coreName;
this.infoSrv = informationServer;
storeRef = new StoreRef(p.getProperty("alfresco.stores", "workspace://SpacesStore"));
batchCount = Integer.parseInt(p.getProperty("alfresco.batch.count", "5000"));
maxLiveSearchers = Integer.parseInt(p.getProperty("alfresco.maxLiveSearchers", "2"));
shardCount = Integer.parseInt(p.getProperty("shard.count", "1"));
shardInstance = Integer.parseInt(p.getProperty("shard.instance", "0"));
shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID);
shardTemplate = p.getProperty("alfresco.template", "");
transformContent = Boolean.parseBoolean(p.getProperty("alfresco.index.transformContent", "true"));
this.trackerStats = this.infoSrv.getTrackerStats();
this.type = type;
this.trackerId = type + "@" + hashCode();
}
/**
* Subclasses must implement behaviour that completes the following steps, in order:
*
* <ol>
* <li>Purge</li>
* <li>Reindex</li>
* <li>Index</li>
* <li>Track repository</li>
* </ol>
*
* @param iterationId an identifier which is uniquely associated with a given iteration.
*/
protected abstract void doTrack(String iterationId) throws Throwable;
private boolean assertTrackerStateRemainsNull() {
/*
* This assertion is added to accommodate DistributedAlfrescoSolrTrackerRaceTest.
* The sleep is needed to allow the test case to add a txn into the queue before
* the tracker makes its call to pull transactions from the test repo client.
*/
try
{
Thread.sleep(5000);
}
catch(Exception e)
{
// Ignore
}
/*
* This ensures that getTrackerState does not have the side effect of setting the
* state instance variable. This allows classes outside of the tracker framework
* to safely call getTrackerState without interfering with the trackers design.
*/
getTrackerState();
return state == null;
}
/**
* Template method - subclasses must implement the {@link Tracker}-specific indexing
* by implementing the abstract method {@link #doTrack(String)}.
*/
@Override
public void track()
{
String iterationId = "IT #" + System.currentTimeMillis();
if(runLock.availablePermits() == 0)
{
LOGGER.info("[{} / {} / {}] Tracker already registered.", coreName, trackerId, iterationId);
return;
}
try
{
/*
* The runLock ensures that for each tracker type (metadata, content, commit, cascade) only one tracker will
* be running at a time.
*/
runLock.acquire();
if (state==null && Boolean.parseBoolean(System.getProperty("alfresco.test", "false")))
{
assert(assertTrackerStateRemainsNull());
}
if(this.state == null)
{
this.state = getTrackerState();
LOGGER.debug("[{} / {} / {}] Global Tracker State set to: {}", coreName, trackerId, iterationId, this.state.toString());
this.state.setRunning(true);
}
else
{
continueState();
this.state.setRunning(true);
}
infoSrv.registerTrackerThread();
try
{
doTrack(iterationId);
}
catch(IndexTrackingShutdownException t)
{
setRollback(true, t);
LOGGER.info("[{} / {} / {}] Tracking cycle stopped. See the stacktrace below for further details.", coreName, trackerId, iterationId, t);
}
catch(Throwable t)
{
setRollback(true, t);
if (t instanceof SocketTimeoutException || t instanceof ConnectException)
{
LOGGER.warn("[{} / {} / {}] Tracking communication timed out. See the stacktrace below for further details.", coreName, trackerId, iterationId);
LOGGER.debug("[{} / {} / {}] Stack trace", coreName, trackerId, iterationId, t);
}
else
{
LOGGER.error("[{} / {} / {}] Tracking failure. See the stacktrace below for further details.", coreName, trackerId, iterationId, t);
}
}
}
catch (InterruptedException e)
{
LOGGER.error("[{} / {} / {}] Semaphore interruption. See the stacktrace below for further details.", coreName, trackerId, iterationId, e);
}
finally
{
infoSrv.unregisterTrackerThread();
ofNullable(state).ifPresent(tstate -> {
// During a rollback state is set to null.
state.setRunning(false);
state.setCheck(false);
});
runLock.release();
}
}
public boolean getRollback()
{
return this.rollback;
}
public Throwable getRollbackCausedBy()
{
return this.rollbackCausedBy;
}
public void setRollback(boolean rollback, Throwable rollbackCausedBy)
{
this.rollback = rollback;
this.rollbackCausedBy = rollbackCausedBy;
}
private void continueState()
{
infoSrv.continueState(state);
state.incrementTrackerCycles();
}
public synchronized void invalidateState()
{
state = null;
}
@Override
public synchronized TrackerState getTrackerState()
{
if(this.state != null)
{
return this.state;
}
else
{
return this.infoSrv.getTrackerInitialState();
}
}
/**
* Allows time for the scheduled asynchronous tasks to complete
*/
synchronized void waitForAsynchronous()
{
AbstractWorkerRunnable currentRunnable = this.threadHandler.peekHeadReindexWorker();
while (currentRunnable != null)
{
checkShutdown();
synchronized (this)
{
try
{
wait(100);
}
catch (InterruptedException e)
{
// Nothing to be done here
}
}
currentRunnable = this.threadHandler.peekHeadReindexWorker();
}
}
int getMaxLiveSearchers()
{
return maxLiveSearchers;
}
void checkShutdown()
{
if(shutdown)
{
throw new IndexTrackingShutdownException();
}
}
@Override
public boolean isAlreadyInShutDownMode()
{
return shutdown;
}
@Override
public void setShutdown(boolean shutdown)
{
this.shutdown = shutdown;
}
@Override
public void shutdown()
{
setShutdown(true);
if(this.threadHandler != null)
{
threadHandler.shutDownThreadPool();
}
}
public Semaphore getWriteLock()
{
return this.writeLock;
}
Semaphore getRunLock()
{
return this.runLock;
}
public Properties getProps()
{
return props;
}
public Type getType()
{
return type;
}
/*
* Copyright (C) 2005-2019 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* 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/>.
*/
package org.alfresco.solr.tracker;
import static java.util.Optional.ofNullable;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.Properties;
import java.util.concurrent.Semaphore;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.solr.IndexTrackingShutdownException;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.TrackerState;
import org.alfresco.solr.client.SOLRAPIClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Abstract base class that provides common {@link Tracker} behaviour.
*
* @author Matt Ward
*/
public abstract class AbstractTracker implements Tracker
{
static final long TIME_STEP_32_DAYS_IN_MS = 1000 * 60 * 60 * 24 * 32L;
static final long TIME_STEP_1_HR_IN_MS = 60 * 60 * 1000L;
static final String SHARD_METHOD_DBID = "DB_ID";
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractTracker.class);
protected Properties props;
protected SOLRAPIClient client;
InformationServer infoSrv;
protected String coreName;
StoreRef storeRef;
long batchCount;
TrackerStats trackerStats;
boolean runPostModelLoadInit = true;
private int maxLiveSearchers;
private volatile boolean shutdown = false;
protected volatile TrackerState state;
protected int shardCount;
protected int shardInstance;
String shardMethod;
protected boolean transformContent;
String shardTemplate;
protected volatile boolean rollback;
/**
* When rollback is set, original error is also gathered in order to provide detailed logging.
*/
protected Throwable rollbackCausedBy;
protected final Type type;
protected final String trackerId;
/**
* Default constructor, strictly for testing.
*/
protected AbstractTracker(Type type)
{
this.type = type;
this.trackerId = type + "@" + hashCode();
}
protected AbstractTracker(Properties p, SOLRAPIClient client, String coreName, InformationServer informationServer,Type type)
{
this.props = p;
this.client = client;
this.coreName = coreName;
this.infoSrv = informationServer;
storeRef = new StoreRef(p.getProperty("alfresco.stores", "workspace://SpacesStore"));
batchCount = Integer.parseInt(p.getProperty("alfresco.batch.count", "5000"));
maxLiveSearchers = Integer.parseInt(p.getProperty("alfresco.maxLiveSearchers", "2"));
shardCount = Integer.parseInt(p.getProperty("shard.count", "1"));
shardInstance = Integer.parseInt(p.getProperty("shard.instance", "0"));
shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID);
shardTemplate = p.getProperty("alfresco.template", "");
transformContent = Boolean.parseBoolean(p.getProperty("alfresco.index.transformContent", "true"));
this.trackerStats = this.infoSrv.getTrackerStats();
this.type = type;
this.trackerId = type + "@" + hashCode();
}
/**
* Subclasses must implement behaviour that completes the following steps, in order:
*
* <ol>
* <li>Purge</li>
* <li>Reindex</li>
* <li>Index</li>
* <li>Track repository</li>
* </ol>
*
* @param iterationId an identifier which is uniquely associated with a given iteration.
*/
protected abstract void doTrack(String iterationId) throws Throwable;
private boolean assertTrackerStateRemainsNull() {
/*
* This assertion is added to accommodate DistributedAlfrescoSolrTrackerRaceTest.
* The sleep is needed to allow the test case to add a txn into the queue before
* the tracker makes its call to pull transactions from the test repo client.
*/
try
{
Thread.sleep(5000);
}
catch(Exception e)
{
// Ignore
}
/*
* This ensures that getTrackerState does not have the side effect of setting the
* state instance variable. This allows classes outside of the tracker framework
* to safely call getTrackerState without interfering with the trackers design.
*/
getTrackerState();
return state == null;
}
/**
* Template method - subclasses must implement the {@link Tracker}-specific indexing
* by implementing the abstract method {@link #doTrack(String)}.
*/
@Override
public void track()
{
String iterationId = "IT #" + System.currentTimeMillis();
if(getRunLock().availablePermits() == 0)
{
LOGGER.info("[{} / {} / {}] Tracker already registered.", coreName, trackerId, iterationId);
return;
}
try
{
/*
* The runLock ensures that for each tracker type (metadata, content, commit, cascade) only one tracker will
* be running at a time.
*/
getRunLock().acquire();
if (state==null && Boolean.parseBoolean(System.getProperty("alfresco.test", "false")))
{
assert(assertTrackerStateRemainsNull());
}
if(this.state == null)
{
this.state = getTrackerState();
LOGGER.debug("[{} / {} / {}] Global Tracker State set to: {}", coreName, trackerId, iterationId, this.state.toString());
this.state.setRunning(true);
}
else
{
continueState();
this.state.setRunning(true);
}
infoSrv.registerTrackerThread();
try
{
doTrack(iterationId);
}
catch(IndexTrackingShutdownException t)
{
setRollback(true, t);
LOGGER.info("[{} / {} / {}] Tracking cycle stopped. See the stacktrace below for further details.", coreName, trackerId, iterationId, t);
}
catch(Throwable t)
{
setRollback(true, t);
if (t instanceof SocketTimeoutException || t instanceof ConnectException)
{
LOGGER.warn("[{} / {} / {}] Tracking communication timed out. See the stacktrace below for further details.", coreName, trackerId, iterationId);
LOGGER.debug("[{} / {} / {}] Stack trace", coreName, trackerId, iterationId, t);
}
else
{
LOGGER.error("[{} / {} / {}] Tracking failure. See the stacktrace below for further details.", coreName, trackerId, iterationId, t);
}
}
}
catch (InterruptedException e)
{
LOGGER.error("[{} / {} / {}] Semaphore interruption. See the stacktrace below for further details.", coreName, trackerId, iterationId, e);
}
finally
{
infoSrv.unregisterTrackerThread();
ofNullable(state).ifPresent(tstate -> {
// During a rollback state is set to null.
state.setRunning(false);
state.setCheck(false);
});
getRunLock().release();
}
}
public boolean getRollback()
{
return this.rollback;
}
public Throwable getRollbackCausedBy()
{
return this.rollbackCausedBy;
}
public void setRollback(boolean rollback, Throwable rollbackCausedBy)
{
this.rollback = rollback;
this.rollbackCausedBy = rollbackCausedBy;
}
private void continueState()
{
infoSrv.continueState(state);
state.incrementTrackerCycles();
}
public synchronized void invalidateState()
{
state = null;
}
@Override
public synchronized TrackerState getTrackerState()
{
if(this.state != null)
{
return this.state;
}
else
{
return this.infoSrv.getTrackerInitialState();
}
}
int getMaxLiveSearchers()
{
return maxLiveSearchers;
}
void checkShutdown()
{
if(shutdown)
{
throw new IndexTrackingShutdownException();
}
}
@Override
public boolean isAlreadyInShutDownMode()
{
return shutdown;
}
@Override
public void setShutdown(boolean shutdown)
{
this.shutdown = shutdown;
}
@Override
public void shutdown()
{
setShutdown(true);
}
/**
* Trackers implementing this method should decide if the Write Lock is applied
* globally for every Tracker Thread (static) or locally for each running Thread
*/
public abstract Semaphore getWriteLock();
/**
* Trackers implementing this method should decide if the Run Lock is applied
* globally for every Tracker Thread (static) or locally for each running Thread
*/
public abstract Semaphore getRunLock();
public Properties getProps()
{
return props;
}
public Type getType()
{
return type;
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractWorkerRunnable.java → search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractWorker.java
+56 -68
View File
@@ -1,68 +1,56 @@
/*
* Copyright (C) 2005-2014 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* 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/>.
*/
package org.alfresco.solr.tracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
abstract class AbstractWorkerRunnable implements Runnable
{
protected final static Logger log = LoggerFactory.getLogger(AbstractWorkerRunnable.class);
QueueHandler queueHandler;
public AbstractWorkerRunnable(QueueHandler qh)
{
this.queueHandler = qh;
}
/*
* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
@Override
public void run()
{
boolean failed = true;
Exception failCausedBy = null;
try
{
doWork();
failed = false;
}
catch (Exception e)
{
log.warn("Index tracking batch hit an unrecoverable error ", e);
failCausedBy = e;
}
finally
{
// Triple check that we get the queue state right
queueHandler.removeFromQueueAndProdHead(this);
if(failed)
{
onFail(failCausedBy);
}
}
}
abstract protected void doWork() throws Exception;
abstract protected void onFail(Throwable failCausedBy);
}
/*
* Copyright (C) 2005-2020 Alfresco Software Limited.
* This file is part of Alfresco
* 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/>.
*/
package org.alfresco.solr.tracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Synchronous Tracking Worker.
*
* @author aborroy
*
*/
public abstract class AbstractWorker
{
protected final static Logger LOGGER = LoggerFactory.getLogger(AbstractWorker.class);
public void run()
{
boolean failed = true;
Exception failCausedBy = null;
try
{
doWork();
failed = false;
}
catch (Exception e)
{
LOGGER.warn("Index tracking batch hit an unrecoverable error ", e);
failCausedBy = e;
}
finally
{
if (failed)
{
onFail(failCausedBy);
}
}
}
abstract protected void doWork() throws Exception;
abstract protected void onFail(Throwable failCausedBy);
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java
+165 -142
View File
@@ -20,13 +20,22 @@ package org.alfresco.solr.tracker;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
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 com.google.common.collect.Lists;
import org.alfresco.error.AlfrescoRuntimeException;
import org.alfresco.httpclient.AuthenticationException;
import org.alfresco.repo.index.shard.ShardMethodEnum;
@@ -56,8 +65,11 @@ public class AclTracker extends AbstractTracker
{
protected final static Logger LOGGER = LoggerFactory.getLogger(AclTracker.class);
private static final int DEFAULT_CHANGE_SET_ACLS_BATCH_SIZE = 100;
private static final int DEFAULT_ACL_BATCH_SIZE = 10;
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 int aclTrackerParallelism;
private int changeSetAclsBatchSize = DEFAULT_CHANGE_SET_ACLS_BATCH_SIZE;
private int aclBatchSize = DEFAULT_ACL_BATCH_SIZE;
@@ -69,6 +81,23 @@ public class AclTracker extends AbstractTracker
private ConcurrentLinkedQueue<Long> aclsToIndex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> aclsToPurge = new ConcurrentLinkedQueue<Long>();
private DocRouter docRouter;
private ForkJoinPool forkJoinPool;
// Share run and write locks across all AclTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
/**
* Default constructor, for testing.
@@ -82,11 +111,18 @@ public class AclTracker extends AbstractTracker
String coreName, InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.ACL);
changeSetAclsBatchSize = Integer.parseInt(p.getProperty("alfresco.changeSetAclsBatchSize", "100"));
aclBatchSize = Integer.parseInt(p.getProperty("alfresco.aclBatchSize", "10"));
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)));
shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID);
docRouter = DocRouterFactory.getRouter(p, ShardMethodEnum.getShardMethod(shardMethod));
threadHandler = new ThreadHandler(p, coreName, "AclTracker");
aclTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.aclTrackerMaxParallelism",
String.valueOf(DEFAULT_ACL_TRACKER_MAX_PARALLELISM)));
forkJoinPool = new ForkJoinPool(aclTrackerParallelism);
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
@Override
@@ -342,7 +378,7 @@ public class AclTracker extends AbstractTracker
AclChangeSet firstChangeSet = firstChangeSets.getAclChangeSets().get(0);
long firstChangeSetCommitTime = firstChangeSet.getCommitTimeMs();
state.setLastGoodChangeSetCommitTimeInIndex(firstChangeSetCommitTime);
setLastChangeSetIdAndCommitTimeInTrackerState(firstChangeSets, state);
setLastChangeSetIdAndCommitTimeInTrackerState(firstChangeSets.getAclChangeSets(), state);
}
}
@@ -384,7 +420,7 @@ public class AclTracker extends AbstractTracker
firstChangeSets = client.getAclChangeSets(null, 0l, null, 2000L, 1);
}
setLastChangeSetIdAndCommitTimeInTrackerState(firstChangeSets, state);
setLastChangeSetIdAndCommitTimeInTrackerState(firstChangeSets.getAclChangeSets(), state);
Long maxChangeSetCommitTimeInRepo = firstChangeSets.getMaxChangeSetCommitTime();
Long maxChangeSetIdInRepo = firstChangeSets.getMaxChangeSetId();
if (maxChangeSetCommitTimeInRepo != null && maxChangeSetIdInRepo != null)
@@ -634,6 +670,15 @@ public class AclTracker extends AbstractTracker
}
/**
* Every ACL Change Set contains a list of ACLs to be indexed.
* This method gets ACL Change Sets from Alfresco Repository to be indexed.
*
* The indexing is performed in batches of ACL Change Sets and the Tracker Status
* is updated in batched of ACLs.
*
* Tracker Status contains the Commit Time from the latest ACL Change Set indexed,
* so new operations can be retrieved from Repository starting with that time.
*
* @throws AuthenticationException
* @throws IOException
* @throws JSONException
@@ -646,10 +691,11 @@ public class AclTracker extends AbstractTracker
boolean upToDate = false;
AclChangeSets aclChangeSets;
BoundedDeque<AclChangeSet> changeSetsFound = new BoundedDeque<AclChangeSet>(100);
HashSet<AclChangeSet> changeSetsIndexed = new LinkedHashSet<AclChangeSet>();
long totalAclCount = 0;
int aclCount = 0;
LOGGER.info("{}-[CORE {}] <init> Tracking ACLs", Thread.currentThread().getId(), coreName);
do
{
try
@@ -664,12 +710,12 @@ public class AclTracker extends AbstractTracker
this.state = getTrackerState();
Long fromCommitTime = getChangeSetFromCommitTime(changeSetsFound, state.getLastGoodChangeSetCommitTimeInIndex());
Long fromCommitTime = getChangeSetFromCommitTime(changeSetsFound,
state.getLastChangeSetCommitTimeOnServer() == 0 ? state.getLastGoodChangeSetCommitTimeInIndex()
: state.getLastChangeSetCommitTimeOnServer());
aclChangeSets = getSomeAclChangeSets(changeSetsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, 2000,
state.getTimeToStopIndexing());
setLastChangeSetIdAndCommitTimeInTrackerState(aclChangeSets, state);
if (aclChangeSets.getAclChangeSets().size() > 0)
{
LOGGER.info("{}-[CORE {}] Found {} ACL change sets after lastTxCommitTime {}, ACL Change Sets from {} to {}",
@@ -686,90 +732,64 @@ public class AclTracker extends AbstractTracker
Thread.currentThread().getId(), coreName, fromCommitTime);
}
ArrayList<AclChangeSet> changeSetBatch = new ArrayList<AclChangeSet>();
for (int i = 0; i < aclChangeSets.getAclChangeSets().size(); i++)
{
AclChangeSet changeSet = aclChangeSets.getAclChangeSets().get(i);
boolean isInIndex = (changeSet.getCommitTimeMs() <= state.getLastIndexedChangeSetCommitTime() &&
infoSrv.aclChangeSetInIndex(changeSet.getId(), true));
if (isInIndex)
{
// Logging progress for large ACL Change Set tracking every 100 tracked ACLs
if (LOGGER.isTraceEnabled())
{
LOGGER.trace("{}-[CORE {}] Tracking {} of {} ACL Change Sets. Change Set Id was already indexed: {}",
Thread.currentThread().getId(), coreName, i + 1, aclChangeSets.getAclChangeSets().size(), changeSet.getId());
}
changeSetsFound.add(changeSet);
}
else
{
// Logging progress for ACL Change Set
if (LOGGER.isTraceEnabled())
{
LOGGER.trace("{}-[CORE {}] Tracking {} of {} ACL Change Sets. Current Change Set Id to be indexed: {}",
Thread.currentThread().getId(), coreName, i + 1, aclChangeSets.getAclChangeSets().size(), changeSet.getId());
}
// Make sure we do not go ahead of where we started - we will check the holes here
// correctly next time
if (changeSet.getCommitTimeMs() > state.getTimeToStopIndexing()) {
upToDate = true;
break;
}
changeSetBatch.add(changeSet);
if (getAclCount(changeSetBatch) > changeSetAclsBatchSize) {
aclCount += indexBatchOfChangeSets(changeSetBatch);
totalAclCount += aclCount;
for (AclChangeSet scheduled : changeSetBatch) {
changeSetsFound.add(scheduled);
changeSetsIndexed.add(scheduled);
// 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;
}
changeSetBatch.clear();
}
}
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()));
if (aclCount > batchCount) {
if (super.infoSrv.getRegisteredSearcherCount() < getMaxLiveSearchers()) {
indexAclChangeSetAfterAsynchronous(changeSetsIndexed, state);
long endElapsed = System.nanoTime();
trackerStats.addElapsedAclTime(aclCount, endElapsed - startElapsed);
startElapsed = endElapsed;
aclCount = 0;
}
}
checkShutdown();
}
if (!changeSetBatch.isEmpty()) {
if (getAclCount(changeSetBatch) > 0) {
aclCount += indexBatchOfChangeSets(changeSetBatch);
totalAclCount += aclCount;
}
for (AclChangeSet scheduled : changeSetBatch) {
changeSetsFound.add(scheduled);
changeSetsIndexed.add(scheduled);
}
changeSetBatch.clear();
}
if(changeSetsIndexed.size() > 0)
// Make sure we do not go ahead of where we started - we will check the holes here
// correctly next time
if (aclChangeSets.getAclChangeSets().stream()
.filter(changeSet -> changeSet.getCommitTimeMs() > state.getTimeToStopIndexing()).findAny().isPresent())
{
indexAclChangeSetAfterAsynchronous(changeSetsIndexed, state);
break;
}
final AtomicInteger counter = new AtomicInteger();
Collection<List<AclChangeSet>> changeSetBatches = aclChangeSets.getAclChangeSets().stream()
.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);
long endElapsed = System.nanoTime();
trackerStats.addElapsedAclTime(aclCount, endElapsed-startElapsed);
startElapsed = endElapsed;
aclCount = 0;
}
};
}
catch(InterruptedException e)
catch(InterruptedException | ExecutionException e)
{
throw new IOException(e);
}
@@ -781,32 +801,43 @@ public class AclTracker extends AbstractTracker
}
while ((aclChangeSets.getAclChangeSets().size() > 0) && (upToDate == false));
LOGGER.info("{}-[CORE {}] Tracked {} ACLs", Thread.currentThread().getId(), coreName, totalAclCount);
LOGGER.info("{}-[CORE {}] <end> Tracked {} ACLs", Thread.currentThread().getId(), coreName, totalAclCount);
}
private void setLastChangeSetIdAndCommitTimeInTrackerState(AclChangeSets aclChangeSets, TrackerState state)
private void setLastChangeSetIdAndCommitTimeInTrackerState(List<AclChangeSet> aclChangeSets, TrackerState state)
{
Long maxChangeSetCommitTime = aclChangeSets.getMaxChangeSetCommitTime();
if(maxChangeSetCommitTime != null)
if (!aclChangeSets.isEmpty())
{
state.setLastChangeSetCommitTimeOnServer(maxChangeSetCommitTime);
}
Long maxChangeSetId = aclChangeSets.getMaxChangeSetId();
if(maxChangeSetId != null)
{
state.setLastChangeSetIdOnServer(maxChangeSetId);
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);
}
}
}
private void indexAclChangeSetAfterAsynchronous(HashSet<AclChangeSet> changeSetsIndexed, TrackerState state)
/**
* Index ACL Change Set transaction after ACLs has been indexed by the worker
*
* @param changeSetsIndexed List of ACL Change Sets indexed
* @param state
* @throws IOException
*/
private void indexAclChangeSetAfterWorker(Collection<AclChangeSet> changeSetsIndexed, TrackerState state)
throws IOException
{
waitForAsynchronous();
for (AclChangeSet set : changeSetsIndexed)
{
super.infoSrv.indexAclTransaction(set, true);
infoSrv.indexAclTransaction(set, true);
// Acl change sets are ordered by commit time and tie-broken by id
if (set.getCommitTimeMs() > state.getLastIndexedChangeSetCommitTime()
|| set.getCommitTimeMs() == state.getLastIndexedChangeSetCommitTime()
@@ -817,8 +848,6 @@ public class AclTracker extends AbstractTracker
}
trackerStats.addChangeSetAcls(set.getAclCount());
}
changeSetsIndexed.clear();
//super.infoSrv.commit();
}
private int getAclCount(List<AclChangeSet> changeSetBatch)
@@ -831,19 +860,24 @@ public class AclTracker extends AbstractTracker
return count;
}
private int indexBatchOfChangeSets(List<AclChangeSet> changeSetBatch) throws AuthenticationException, IOException, JSONException
{
int aclCount = 0;
ArrayList<AclChangeSet> nonEmptyChangeSets = new ArrayList<AclChangeSet>(changeSetBatch.size());
for (AclChangeSet set : changeSetBatch)
{
if (set.getAclCount() > 0)
{
nonEmptyChangeSets.add(set);
}
}
/**
* Index ACLs from ACL Change Sets contained in changeSetBatch
* When total ACL indexed count is greater than the specified for a single execution
* (maxAclsPerExecution), no more ACL Change Sets are processed.
*
* @param changeSetBatch List of ACL Change Sets to be indexed
* @return List of ACL Change Set indexed and Count of ACL indexed
*
* @throws AuthenticationException
* @throws IOException
* @throws JSONException
*/
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)
.collect(Collectors.toList());
ArrayList<Acl> aclBatch = new ArrayList<Acl>();
List<Acl> acls = client.getAcls(nonEmptyChangeSets, null, Integer.MAX_VALUE);
if (LOGGER.isDebugEnabled())
@@ -851,40 +885,29 @@ public class AclTracker extends AbstractTracker
LOGGER.debug("{}-[CORE {}] Found {} Acls from Acl Change Sets: {}", Thread.currentThread().getId(),
coreName, acls.size(), nonEmptyChangeSets);
}
List<List<Acl>> aclBatches = Lists.partition(acls, aclBatchSize);
for (Acl acl : acls)
{
if (LOGGER.isTraceEnabled())
{
LOGGER.trace("{}-[CORE {}] Adding ACL {} to scheduled indexing job", Thread.currentThread().getId(),
coreName, acl.toString());
}
aclBatch.add(acl);
if (aclBatch.size() > aclBatchSize)
{
aclCount += aclBatch.size();
AclIndexWorkerRunnable aiwr = new AclIndexWorkerRunnable(this.threadHandler, aclBatch);
this.threadHandler.scheduleTask(aiwr);
aclBatch = new ArrayList<Acl>();
}
}
if (aclBatch.size() > 0)
{
aclCount += aclBatch.size();
AclIndexWorkerRunnable aiwr = new AclIndexWorkerRunnable(this.threadHandler, aclBatch);
this.threadHandler.scheduleTask(aiwr);
aclBatch = new ArrayList<Acl>();
}
return aclCount;
Integer processedAcls = forkJoinPool.submit(() ->
aclBatches.parallelStream().map(batch -> {
new AclIndexWorker(batch).run();
return batch.size();
}).reduce(0, Integer::sum)
).get();
return processedAcls;
}
class AclIndexWorkerRunnable extends AbstractWorkerRunnable
/**
* ACL Indexer
*/
class AclIndexWorker extends AbstractWorker
{
List<Acl> acls;
AclIndexWorkerRunnable(QueueHandler queueHandler, List<Acl> acls)
AclIndexWorker(List<Acl> acls)
{
super(queueHandler);
this.acls = acls;
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java
+94 -44
View File
@@ -25,11 +25,16 @@ import static org.alfresco.solr.utils.Utils.notNullOrEmpty;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Semaphore;
import com.google.common.collect.Lists;
import org.alfresco.httpclient.AuthenticationException;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.client.NodeMetaData;
@@ -45,14 +50,46 @@ import org.slf4j.LoggerFactory;
*/
public class CascadeTracker extends AbstractTracker implements Tracker
{
private static final Logger LOGGER = LoggerFactory.getLogger(CascadeTracker.class);
protected final static Logger LOGGER = LoggerFactory.getLogger(CascadeTracker.class);
private static final int DEFAULT_CASCADE_TRACKER_MAX_PARALLELISM = 32;
private static final int DEFAULT_CASCADE_NODE_BATCH_SIZE = 10;
// Share run and write locks across all CascadeTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
private int cascadeBatchSize;
private ForkJoinPool forkJoinPool;
private int cascadeTrackerParallelism;
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
public CascadeTracker(Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.CASCADE);
threadHandler = new ThreadHandler(p, coreName, "CascadeTracker");
cascadeTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.cascadeTrackerMaxParallelism",
String.valueOf(DEFAULT_CASCADE_TRACKER_MAX_PARALLELISM)));
cascadeBatchSize = Integer.parseInt(p.getProperty("alfresco.cascadeNodeBatchSize",
String.valueOf(DEFAULT_CASCADE_NODE_BATCH_SIZE)));;
forkJoinPool = new ForkJoinPool(cascadeTrackerParallelism);
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
CascadeTracker()
@@ -85,24 +122,22 @@ public class CascadeTracker extends AbstractTracker implements Tracker
processCascades(iterationId);
}
private void updateTransactionsAfterAsynchronous(List<Transaction> txsIndexed)
private void updateTransactionsAfterWorker(List<Transaction> txsIndexed)
throws IOException
{
waitForAsynchronous();
for (Transaction tx : txsIndexed)
{
super.infoSrv.updateTransaction(tx);
}
}
class CascadeIndexWorkerRunnable extends AbstractWorkerRunnable
class CascadeIndexWorker extends AbstractWorker
{
InformationServer infoServer;
List<NodeMetaData> nodes;
CascadeIndexWorkerRunnable(QueueHandler queueHandler, List<NodeMetaData> nodes, InformationServer infoServer)
CascadeIndexWorker(List<NodeMetaData> nodes, InformationServer infoServer)
{
super(queueHandler);
this.infoServer = infoServer;
this.nodes = nodes;
}
@@ -116,7 +151,7 @@ public class CascadeTracker extends AbstractTracker implements Tracker
@Override
protected void onFail(Throwable failCausedBy)
{
setRollback(true, failCausedBy);
setRollback(true, failCausedBy);
}
}
@@ -129,58 +164,69 @@ public class CascadeTracker extends AbstractTracker implements Tracker
private void processCascades(String iterationId) throws IOException
{
int num = 50;
List<Transaction> txBatch;
do
{
try
{
List<Transaction> txBatch = null;
long totalUpdatedDocs = 0;
do {
try {
getWriteLock().acquire();
txBatch = infoSrv.getCascades(num);
if(txBatch.size() == 0)
if (txBatch.size() > 0)
{
//No transactions to process for cascades.
LOGGER.info("{}-[CORE {}] Found {} transactions, transactions from {} to {}",
Thread.currentThread().getId(),
coreName,
txBatch.size(),
txBatch.get(0),
txBatch.get(txBatch.size() - 1));
}
else
{
LOGGER.info("{}-[CORE {}] No transaction found",
Thread.currentThread().getId(), coreName);
}
if(txBatch.size() == 0) {
return;
}
ArrayList<Long> txIds = new ArrayList<>();
Set<Long> txIdSet = new HashSet<>();
for (Transaction tx : txBatch)
{
for (Transaction tx : txBatch) {
txIds.add(tx.getId());
txIdSet.add(tx.getId());
}
List<NodeMetaData> nodeMetaDatas = infoSrv.getCascadeNodes(txIds);
Integer processedCascades = 0;
if(nodeMetaDatas.size() > 0)
{
LinkedList<NodeMetaData> stack = new LinkedList<>();
stack.addAll(nodeMetaDatas);
int batchSize = 10;
if(nodeMetaDatas.size() > 0) {
List<List<NodeMetaData>> nodeBatches = Lists.partition(nodeMetaDatas, cascadeBatchSize);
do
{
List<NodeMetaData> batch = new ArrayList<>();
while (batch.size() < batchSize && stack.size() > 0)
{
batch.add(stack.removeFirst());
}
processedCascades = forkJoinPool.submit( () ->
nodeBatches.parallelStream().map( batch -> {
CascadeIndexWorker worker = new CascadeIndexWorker(batch, infoSrv);
worker.run();
if (LOGGER.isTraceEnabled())
{
String nodes = notNullOrEmpty(batch).stream()
.map(NodeMetaData::getId)
.map(Object::toString)
.collect(joining(","));
LOGGER.trace("[{} / {} / {} / {}] Worker has been created for nodes {}", coreName, trackerId, iterationId, worker.hashCode(), nodes);
}
return batch.size();
}).reduce(0, Integer::sum)
).get();
CascadeIndexWorkerRunnable worker = new CascadeIndexWorkerRunnable(this.threadHandler, batch, infoSrv);
if (LOGGER.isTraceEnabled())
{
String nodes = notNullOrEmpty(batch).stream()
.map(NodeMetaData::getId)
.map(Object::toString)
.collect(joining(","));
LOGGER.trace("[{} / {} / {} / {}] Worker has been created for nodes {}", coreName, trackerId, iterationId, worker.hashCode(), nodes);
}
this.threadHandler.scheduleTask(worker);
} while (stack.size() > 0);
}
//Update the transaction records.
updateTransactionsAfterAsynchronous(txBatch);
updateTransactionsAfterWorker(txBatch);
totalUpdatedDocs += processedCascades;
}
catch (AuthenticationException e)
{
@@ -193,11 +239,15 @@ public class CascadeTracker extends AbstractTracker implements Tracker
catch(InterruptedException e)
{
throw new IOException(e);
}
finally
} catch (ExecutionException e) {
e.printStackTrace();
} finally
{
getWriteLock().release();
}
} while(txBatch.size() > 0);
LOGGER.info("{}-[CORE {}] Updated {} DOCs", Thread.currentThread().getId(), coreName, totalUpdatedDocs);
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java
+24 -4
View File
@@ -23,8 +23,11 @@ import static java.util.Optional.empty;
import static java.util.Optional.ofNullable;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import org.alfresco.solr.InformationServer;
@@ -49,7 +52,21 @@ public class CommitTracker extends AbstractTracker
private Optional<CascadeTracker> cascadeTracker = empty();
private AtomicInteger rollbackCount = new AtomicInteger(0);
protected final static Logger log = LoggerFactory.getLogger(CommitTracker.class);
protected final static Logger LOGGER = LoggerFactory.getLogger(CommitTracker.class);
// Share run and write locks across all CommitTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
/**
* Default constructor, for testing.
@@ -83,6 +100,9 @@ public class CommitTracker extends AbstractTracker
commitInterval = Long.parseLong(p.getProperty("alfresco.commitInterval", "60000")); // Default: commit once per minute
newSearcherInterval = Integer.parseInt(p.getProperty("alfresco.newSearcherInterval", "120000")); // Default: Open searchers every two minutes
lastSearcherOpened = lastCommit = System.currentTimeMillis();
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
public boolean hasMaintenance() throws Exception
@@ -198,17 +218,17 @@ public class CommitTracker extends AbstractTracker
// Log reasons why the rollback is performed
if (aclTracker.getRollbackCausedBy() != null)
{
log.warn("Rollback performed due to ACL Tracker error", aclTracker.getRollbackCausedBy());
LOGGER.warn("Rollback performed due to ACL Tracker error", aclTracker.getRollbackCausedBy());
}
if (metadataTracker.getRollbackCausedBy() != null)
{
log.warn("Rollback performed due to Metadata Tracker error", metadataTracker.getRollbackCausedBy());
LOGGER.warn("Rollback performed due to Metadata Tracker error", metadataTracker.getRollbackCausedBy());
}
}
catch (Exception e)
{
log.error("Rollback failed", e);
LOGGER.error("Rollback failed", e);
}
finally
{
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java
+64 -43
View File
@@ -18,16 +18,21 @@
*/
package org.alfresco.solr.tracker;
import java.io.IOException;
import java.util.List;
import java.util.Properties;
import com.google.common.collect.Lists;
import org.alfresco.solr.AlfrescoSolrDataModel.TenantAclIdDbId;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.client.SOLRAPIClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Semaphore;
/**
* This tracker queries for docs with unclean content, and then updates them.
* Similar to org.alfresco.repo.search.impl.lucene.ADMLuceneIndexerImpl
@@ -37,18 +42,42 @@ import org.slf4j.LoggerFactory;
public class ContentTracker extends AbstractTracker implements Tracker
{
protected final static Logger log = LoggerFactory.getLogger(ContentTracker.class);
private int contentReadBatchSize;
protected final static Logger LOGGER = LoggerFactory.getLogger(ContentTracker.class);
private static int DEFAULT_CONTENT_UPDATE_BATCH_SIZE = 2000;
private static final int DEFAULT_CONTENT_TRACKER_MAX_PARALLELISM = 32;
private int contentTrackerParallelism;
private int contentUpdateBatchSize;
// Share run and write locks across all ContentTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
private ForkJoinPool forkJoinPool;
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
public ContentTracker(Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.CONTENT);
contentReadBatchSize = Integer.parseInt(p.getProperty("alfresco.contentReadBatchSize", "100"));
contentUpdateBatchSize = Integer.parseInt(p.getProperty("alfresco.contentUpdateBatchSize", "1000"));
threadHandler = new ThreadHandler(p, coreName, "ContentTracker");
contentUpdateBatchSize = Integer.parseInt(p.getProperty("alfresco.contentUpdateBatchSize",
String.valueOf(DEFAULT_CONTENT_UPDATE_BATCH_SIZE)));
contentTrackerParallelism = Integer.parseInt(p.getProperty("alfresco.contentTrackerMaxParallelism",
String.valueOf(DEFAULT_CONTENT_TRACKER_MAX_PARALLELISM)));
forkJoinPool = new ForkJoinPool(contentTrackerParallelism);
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
ContentTracker()
@@ -59,51 +88,44 @@ public class ContentTracker extends AbstractTracker implements Tracker
@Override
protected void doTrack(String iterationId) throws Exception
{
//System.out.println("############## Content Tracker doTrack()");
try {
try
{
long startElapsed = System.nanoTime();
checkShutdown();
final int ROWS = contentReadBatchSize;
int start = 0;
long totalDocs = 0l;
checkShutdown();
while (true) {
try
{
getWriteLock().acquire();
List<TenantAclIdDbId> docs = this.infoSrv.getDocsWithUncleanContent(start, ROWS);
//System.out.println("####################### Unclean content: "+docs.size()+" ##############################:"+totalDocs);
List<TenantAclIdDbId> docs = this.infoSrv.getDocsWithUncleanContent();
if (docs.size() == 0) {
break;
}
int docsUpdatedSinceLastCommit = 0;
for (TenantAclIdDbId doc : docs) {
ContentIndexWorkerRunnable ciwr = new ContentIndexWorkerRunnable(super.threadHandler, doc, infoSrv);
super.threadHandler.scheduleTask(ciwr);
docsUpdatedSinceLastCommit++;
List<List<TenantAclIdDbId>> docBatches = Lists.partition(docs, contentUpdateBatchSize);
for (List<TenantAclIdDbId> batch : docBatches) {
if (docsUpdatedSinceLastCommit >= contentUpdateBatchSize) {
super.waitForAsynchronous();
checkShutdown();
//this.infoSrv.commit();
long endElapsed = System.nanoTime();
trackerStats.addElapsedContentTime(docsUpdatedSinceLastCommit, endElapsed - startElapsed);
startElapsed = endElapsed;
docsUpdatedSinceLastCommit = 0;
}
}
if (docsUpdatedSinceLastCommit > 0) {
super.waitForAsynchronous();
checkShutdown();
//this.infoSrv.commit();
Integer processedDocuments = forkJoinPool.submit(() ->
// Parallel task here, for example
batch.parallelStream().map(doc -> {
ContentIndexWorkerRunnable ciwr = new ContentIndexWorkerRunnable(doc, infoSrv);
ciwr.run();
return 1;
}).reduce(0, Integer::sum)
).get();
long endElapsed = System.nanoTime();
trackerStats.addElapsedContentTime(docsUpdatedSinceLastCommit, endElapsed - startElapsed);
}
trackerStats.addElapsedContentTime(processedDocuments, endElapsed - startElapsed);
startElapsed = endElapsed;
};
totalDocs += docs.size();
checkShutdown();
}
@@ -113,7 +135,8 @@ public class ContentTracker extends AbstractTracker implements Tracker
}
}
log.info("total number of docs with content updated: " + totalDocs);
LOGGER.info("{}-[CORE {}] Total number of docs with content updated: {} ", Thread.currentThread().getId(), coreName, totalDocs);
}
catch(Exception e)
{
@@ -134,14 +157,13 @@ public class ContentTracker extends AbstractTracker implements Tracker
this.infoSrv.setCleanContentTxnFloor(-1);
}
class ContentIndexWorkerRunnable extends AbstractWorkerRunnable
class ContentIndexWorkerRunnable extends AbstractWorker
{
InformationServer infoServer;
TenantAclIdDbId doc;
ContentIndexWorkerRunnable(QueueHandler queueHandler, TenantAclIdDbId doc, InformationServer infoServer)
ContentIndexWorkerRunnable(TenantAclIdDbId doc, InformationServer infoServer)
{
super(queueHandler);
this.doc = doc;
this.infoServer = infoServer;
}
@@ -150,7 +172,6 @@ public class ContentTracker extends AbstractTracker implements Tracker
protected void doWork() throws Exception
{
checkShutdown();
//System.out.println("################ Update doc:"+doc.dbId);
this.infoServer.updateContentToIndexAndCache(doc.dbId, doc.tenant);
}
@@ -158,7 +179,7 @@ public class ContentTracker extends AbstractTracker implements Tracker
protected void onFail(Throwable failCausedBy)
{
// This will be redone in future tracking operations
log.warn("Content tracker failed due to {}", failCausedBy.getMessage(), failCausedBy);
LOGGER.warn("Content tracker failed due to {}", failCausedBy.getMessage(), failCausedBy);
}
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java
+386 -265
View File
@@ -18,17 +18,7 @@
*/
package org.alfresco.solr.tracker;
import static org.alfresco.repo.index.shard.ShardMethodEnum.DB_ID_RANGE;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
import com.google.common.collect.Lists;
import org.alfresco.error.AlfrescoRuntimeException;
import org.alfresco.httpclient.AuthenticationException;
import org.alfresco.repo.index.shard.ShardState;
@@ -49,17 +39,41 @@ import org.json.JSONException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;
import static org.alfresco.repo.index.shard.ShardMethodEnum.DB_ID_RANGE;
/*
* This tracks two things: transactions and metadata nodes
* @author Ahmed Owian
*/
public class MetadataTracker extends CoreStatePublisher implements Tracker
{
protected final static Logger log = LoggerFactory.getLogger(MetadataTracker.class);
private static final int DEFAULT_TRANSACTION_DOCS_BATCH_SIZE = 100;
private static final int DEFAULT_NODE_BATCH_SIZE = 10;
private int transactionDocsBatchSize = DEFAULT_TRANSACTION_DOCS_BATCH_SIZE;
private int nodeBatchSize = DEFAULT_NODE_BATCH_SIZE;
protected final static Logger LOGGER = LoggerFactory.getLogger(MetadataTracker.class);
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 int matadataTrackerParallelism;
private int transactionDocsBatchSize;
private int nodeBatchSize;
private int maxNumberOfTransactions;
private ConcurrentLinkedQueue<Long> transactionsToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToPurge = new ConcurrentLinkedQueue<>();
@@ -67,6 +81,21 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
private ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<>();
private ForkJoinPool forkJoinPool;
// Share run and write locks across all MetadataTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
/**
* Check if nextTxCommitTimeService is available in the repository.
@@ -115,12 +144,22 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
InformationServer informationServer, boolean checkRepoServicesAvailability)
{
super(isMaster, p, client, coreName, informationServer, Tracker.Type.METADATA);
transactionDocsBatchSize = Integer.parseInt(p.getProperty("alfresco.transactionDocsBatchSize", "100"));
nodeBatchSize = Integer.parseInt(p.getProperty("alfresco.nodeBatchSize", "10"));
threadHandler = new ThreadHandler(p, coreName, "MetadataTracker");
transactionDocsBatchSize = Integer.parseInt(p.getProperty("alfresco.transactionDocsBatchSize",
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",
String.valueOf(DEFAULT_METADATA_TRACKER_MAX_PARALLELISM)));
String[] minTxninitialRangeString = p.getProperty("solr.initial.transaction.range", DEFAULT_INITIAL_TRANSACTION_RANGE).split("-");
cascadeTrackerEnabled = informationServer.cascadeTrackingEnabled();
String[] minTxninitialRangeString = p.getProperty("solr.initial.transaction.range", "0-2000").split("-");
minTxnIdRange = new Pair<Long, Long>(Long.valueOf(minTxninitialRangeString[0]), Long.valueOf(minTxninitialRangeString[1]));
minTxnIdRange = new Pair<>(Long.valueOf(minTxninitialRangeString[0]), Long.valueOf(minTxninitialRangeString[1]));
forkJoinPool = new ForkJoinPool(matadataTrackerParallelism);
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
// In order to apply performance optimizations, checking the availability of Repo Web Scripts is required.
// As these services are available from ACS 6.2
@@ -134,11 +173,11 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
}
catch (NoSuchMethodException e)
{
log.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)
{
log.error("Checking nextTxCommitTimeService failed.", e);
LOGGER.error("Checking nextTxCommitTimeService failed.", e);
}
// Try invoking txIntervalCommitTime service
@@ -151,12 +190,12 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
}
catch (NoSuchMethodException e)
{
log.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)
{
log.error("Checking txIntervalCommitTimeServiceAvailable failed.", e);
LOGGER.error("Checking txIntervalCommitTimeServiceAvailable failed.", e);
}
}
}
@@ -171,7 +210,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
@Override
protected void doTrack(String iterationId) throws AuthenticationException, IOException, JSONException, EncoderException
{
log.debug("### MetadataTracker doTrack ###");
// MetadataTracker must wait until ModelTracker has run
ModelTracker modelTracker = this.infoSrv.getAdminHandler().getTrackerRegistry().getModelTracker();
if (modelTracker != null && modelTracker.hasModels())
@@ -208,13 +246,11 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
private void trackRepository() throws IOException, AuthenticationException, JSONException, EncoderException
{
log.debug("####### MetadataTracker trackRepository Start #######");
checkShutdown();
// Check we are tracking the correct repository
TrackerState state = super.getTrackerState();
log.debug("####### MetadataTracker check CYCLE #######");
log.debug(String.format("%s ### state: %s", coreName, state.toString()));
if(state.getTrackerCycles() == 0)
{
//We have a new tracker state so do the checks.
@@ -254,7 +290,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
{
state.setCheckedLastTransactionTime(true);
state.setCheckedFirstTransactionTime(true);
log.info("No transactions found - no verification required");
LOGGER.info("No transactions found - no verification required");
firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), null, minTxnIdRange.getSecond(), 1);
if (!firstTransactions.getTransactions().isEmpty())
@@ -262,7 +298,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
Transaction firstTransaction = firstTransactions.getTransactions().get(0);
long firstTransactionCommitTime = firstTransaction.getCommitTimeMs();
state.setLastGoodTxCommitTimeInIndex(firstTransactionCommitTime);
setLastTxCommitTimeAndTxIdInTrackerState(firstTransactions, state);
setLastTxCommitTimeAndTxIdInTrackerState(firstTransactions);
}
}
@@ -286,7 +322,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
}
catch (NoSuchMethodException e)
{
log.warn("txIntervalCommitTimeServiceAvailable is not available. If you are using DB_ID_RANGE shard method, "
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());
}
}
@@ -305,20 +341,20 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
if (setSize == 0)
{
log.error("First transaction was not found with the correct timestamp.");
log.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match.");
log.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database.");
log.error("You can also check your SOLR connection details in solrcore.properties.");
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("You can also check your SOLR connection details in solrcore.properties.");
throw new AlfrescoRuntimeException("Initial transaction not found with correct timestamp");
}
else if (setSize == 1)
{
state.setCheckedFirstTransactionTime(true);
log.info("Verified first transaction and timestamp in index");
LOGGER.info("Verified first transaction and timestamp in index");
}
else
{
log.warn("Duplicate initial transaction found with correct timestamp");
LOGGER.warn("Duplicate initial transaction found with correct timestamp");
}
}
}
@@ -332,7 +368,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
firstTransactions = client.getTransactions(null, minTxnIdRange.getFirst(), null, minTxnIdRange.getSecond(), 1);
}
setLastTxCommitTimeAndTxIdInTrackerState(firstTransactions, state);
setLastTxCommitTimeAndTxIdInTrackerState(firstTransactions);
Long maxTxnCommitTimeInRepo = firstTransactions.getMaxTxnCommitTime();
Long maxTxnIdInRepo = firstTransactions.getMaxTxnId();
if (maxTxnCommitTimeInRepo != null && maxTxnIdInRepo != null)
@@ -340,19 +376,19 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
Transaction maxTxInIndex = this.infoSrv.getMaxTransactionIdAndCommitTimeInIndex();
if (maxTxInIndex.getCommitTimeMs() > maxTxnCommitTimeInRepo)
{
log.error("Last transaction was found in index with timestamp later than that of repository.");
log.error("Max Tx In Index: " + maxTxInIndex.getId() + ", In Repo: " + maxTxnIdInRepo);
log.error("Max Tx Commit Time In Index: " + maxTxInIndex.getCommitTimeMs() + ", In Repo: "
LOGGER.error("Last transaction was found in index with timestamp later than that of repository.");
LOGGER.error("Max Tx In Index: " + maxTxInIndex.getId() + ", In Repo: " + maxTxnIdInRepo);
LOGGER.error("Max Tx Commit Time In Index: " + maxTxInIndex.getCommitTimeMs() + ", In Repo: "
+ maxTxnCommitTimeInRepo);
log.error("SOLR has successfully connected to your repository however the SOLR indexes and repository database do not match.");
log.error("If this is a new or rebuilt database your SOLR indexes also need to be re-built to match the database.");
log.error("You can also check your SOLR connection details in solrcore.properties.");
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");
}
else
{
state.setCheckedLastTransactionTime(true);
log.info("Verified last transaction timestamp in index less than or equal to that of repository.");
LOGGER.info("Verified last transaction timestamp in index less than or equal to that of repository.");
}
}
}
@@ -390,9 +426,9 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
for (Node node : nodes)
{
docCount++;
if (log.isDebugEnabled())
if (LOGGER.isDebugEnabled())
{
log.debug(node.toString());
LOGGER.debug(node.toString());
}
this.infoSrv.indexNode(node, false);
checkShutdown();
@@ -400,14 +436,15 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
// Index the transaction doc after the node - if this is not found then a reindex will be done.
this.infoSrv.indexTransaction(info, false);
log.info("INDEX ACTION - Transaction {} has been indexed", transactionId);
LOGGER.info("INDEX ACTION - Transaction {} has been indexed", transactionId);
requiresCommit = true;
trackerStats.addTxDocs(nodes.size());
}
else
{
log.info("INDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", transactionId);
LOGGER.info("INDEX ACTION - Transaction {} was not found in database, it has NOT been reindexed", transactionId);
}
}
@@ -416,7 +453,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
if(this.infoSrv.getRegisteredSearcherCount() < getMaxLiveSearchers())
{
checkShutdown();
//this.infoSrv.commit();
long endElapsed = System.nanoTime();
trackerStats.addElapsedNodeTime(docCount, endElapsed-startElapsed);
startElapsed = endElapsed;
@@ -448,7 +484,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
node.setTxnId(Long.MAX_VALUE);
this.infoSrv.indexNode(node, false);
log.info("INDEX ACTION - Node {} has been reindexed", node.getId());
LOGGER.info("INDEX ACTION - Node {} has been reindexed", node.getId());
requiresCommit = true;
}
checkShutdown();
@@ -489,9 +525,9 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
for (Node node : nodes)
{
docCount++;
if (log.isDebugEnabled())
if (LOGGER.isDebugEnabled())
{
log.debug(node.toString());
LOGGER.debug(node.toString());
}
this.infoSrv.indexNode(node, true);
checkShutdown();
@@ -499,11 +535,11 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
// Index the transaction doc after the node - if this is not found then a reindex will be done.
this.infoSrv.indexTransaction(info, true);
log.info("REINDEX ACTION - Transaction {} has been reindexed", transactionId);
LOGGER.info("REINDEX ACTION - Transaction {} has been reindexed", transactionId);
}
else
{
log.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);
}
}
@@ -546,7 +582,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
node.setTxnId(Long.MAX_VALUE);
this.infoSrv.indexNode(node, true);
log.info("REINDEX ACTION - Node {} has been reindexed", node.getId());
LOGGER.info("REINDEX ACTION - Node {} has been reindexed", node.getId());
requiresCommit = true;
}
checkShutdown();
@@ -568,7 +604,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
if (query != null)
{
this.infoSrv.reindexNodeByQuery(query);
log.info("REINDEX ACTION - Nodes from query {} have been reindexed", query);
LOGGER.info("REINDEX ACTION - Nodes from query {} have been reindexed", query);
requiresCommit = true;
}
checkShutdown();
@@ -593,7 +629,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
// make sure it is cleaned out so we do not miss deletes
this.infoSrv.deleteByTransactionId(transactionId);
requiresCommit = true;
log.info("PURGE ACTION - Purged transactionId {}", transactionId);
LOGGER.info("PURGE ACTION - Purged transactionId {}", transactionId);
}
checkShutdown();
}
@@ -601,7 +637,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
if(requiresCommit)
{
checkShutdown();
//this.infoSrv.commit();
}
}
@@ -614,12 +649,34 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
{
// make sure it is cleaned out so we do not miss deletes
this.infoSrv.deleteByNodeId(nodeId);
log.info("PURGE ACTION - Purged nodeId {}", nodeId);
LOGGER.info("PURGE ACTION - Purged nodeId {}", nodeId);
}
checkShutdown();
}
}
/**
* The fromCommitTime tells getSomeTransactions() where to start, this actually fairly straight forward.
*
* What makes this code so tricky to understand is the state.getTimeToStopIndexing().
*
* There are two scenarios to keep in mind:
*
* 1) Full re-index: In this scenario the state.getTimeToStopIndexing() will never stop the indexing.
*
* 2) Up-to-date indexing: This is where state.getTimeToStopIndexing() gets interesting. In this scenario
* the Solr index is already up to date with the repo and it is tracking new transactions. The state.getTimeToStopIndexing()
* in this scenario causes the getSomeTransactions() call to stop returning results if it finds a transaction
* beyond a specific point in time. This will break out of this loop and end the tracker run.
*
* The next time the metadata tracker runs the "continueState()" method applies the "hole retention"
* to state.getLastGoodTxCommitTimeInIndex(). This causes the state.getLastGoodTxCommitTimeInIndex() to scan
* for prior transactions that might have been missed.
*
* @param txnsFound
* @param lastGoodTxCommitTimeInIndex
* @return
*/
protected Long getTxFromCommitTime(BoundedDeque<Transaction> txnsFound, long lastGoodTxCommitTimeInIndex) {
if (txnsFound.size() > 0)
{
@@ -662,10 +719,6 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
// step forward in time until we find something or hit the time bound
// max id unbounded
Long startTime = fromCommitTime == null ? 0L : fromCommitTime;
log.debug(String.format("#### %s MetadataTracker getSomeTransactions start time: %d end: %d",
this.coreName,
startTime,
endTime));
if(startTime == 0)
{
return client.getTransactions(startTime,
@@ -675,6 +728,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
maxResults,
shardstate);
}
do
{
transactions = client.getTransactions(startTime, null, startTime + actualTimeStep, null, maxResults, shardstate);
@@ -686,7 +740,8 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
Long nextTxCommitTime = client.getNextTxCommitTime(coreName, startTime);
if (nextTxCommitTime != -1)
{
log.info("Advancing transactions from {} to {}", startTime, nextTxCommitTime);
LOGGER.info("{}-[CORE {}] Advancing transactions from {} to {}",
Thread.currentThread().getId(), coreName, startTime, nextTxCommitTime);
transactions = client.getTransactions(nextTxCommitTime, null, nextTxCommitTime + actualTimeStep, null, maxResults, shardstate);
}
}
@@ -697,7 +752,141 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
return transactions;
}
/**
* When using DB_ID_RANGE, fromCommitTime cannot be before the commit time of the first transaction
* for the DB_ID_RANGE to be indexed and commit time of the last transaction cannot be lower than fromCommitTime.
* When there isn't nodes in that range, -1 is returned as commit times
*
* @param fromCommitTime Starting commit time to get transactions from Repository
* @param txnsFound List of transactions previously found
* @return List of transactions to be indexed
*
* @throws NoSuchMethodException
* @throws AuthenticationException
* @throws IOException
* @throws JSONException
* @throws EncoderException
*/
private Transactions getDBIDRangeTransactions(Long fromCommitTime, BoundedDeque<Transaction> txnsFound)
throws NoSuchMethodException, AuthenticationException, IOException, JSONException, EncoderException
{
boolean shardOutOfRange = false;
DBIDRangeRouter dbIdRangeRouter = (DBIDRangeRouter) docRouter;
Pair<Long, Long> commitTimes = client.getTxIntervalCommitTime(coreName,
dbIdRangeRouter.getStartRange(), dbIdRangeRouter.getEndRange());
Long shardMinCommitTime = commitTimes.getFirst();
Long shardMaxCommitTime = commitTimes.getSecond();
// Node Range it's not still available in repository
if (shardMinCommitTime == -1)
{
LOGGER.debug(
"{}-[CORE {}] [DB_ID_RANGE] No nodes in range [{}-{}] "
+ "exist in the repository. Indexing only latest transaction.",
Thread.currentThread().getId(), coreName, dbIdRangeRouter.getStartRange(),
dbIdRangeRouter.getEndRange());
shardOutOfRange = true;
}
if (fromCommitTime > shardMaxCommitTime)
{
LOGGER.debug(
"{}-[CORE {}] [DB_ID_RANGE] Last commit time is greater that max commit time in in range [{}-{}]. "
+ "Indexing only latest transaction.",
Thread.currentThread().getId(), coreName, dbIdRangeRouter.getStartRange(),
dbIdRangeRouter.getEndRange());
shardOutOfRange = true;
}
// Initial commit time for Node Range is greater than calculated from commit time
if (fromCommitTime < shardMinCommitTime)
{
LOGGER.debug("{}-[CORE {}] [DB_ID_RANGE] Skipping transactions from {} to {}",
Thread.currentThread().getId(), coreName, fromCommitTime, shardMinCommitTime);
fromCommitTime = shardMinCommitTime;
}
Transactions transactions = getSomeTransactions(txnsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions,
state.getTimeToStopIndexing());
// When transactions are out of Shard range, only the latest transaction needs to be indexed
// in order to preserve the state up-to-date of the MetadataTracker
if (shardOutOfRange)
{
Transaction latestTransaction = new Transaction();
latestTransaction.setCommitTimeMs(transactions.getMaxTxnCommitTime());
latestTransaction.setId(transactions.getMaxTxnId());
transactions = new Transactions(
Arrays.asList(latestTransaction),
transactions.getMaxTxnCommitTime(),
transactions.getMaxTxnId());
}
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)
{
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()));
}
//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
{
long startElapsed = System.nanoTime();
@@ -705,19 +894,19 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
boolean upToDate = false;
Transactions transactions;
BoundedDeque<Transaction> txnsFound = new BoundedDeque<Transaction>(100);
HashSet<Transaction> txsIndexed = new LinkedHashSet<>();
long totalUpdatedDocs = 0;
int docCount = 0;
int totalUpdatedDocs = 0;
LOGGER.info("{}-[CORE {}] Starting metadata tracker execution", Thread.currentThread().getId(), coreName);
do
{
try
{
/*
* This write lock is used to lock out the Commit Tracker. The ensures that the MetaDataTracker will
* not be indexing content while commits or rollbacks are occurring.
*/
getWriteLock().acquire();
/*
@@ -725,184 +914,94 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
* tracker state could have been invalidated due to a rollback by the CommitTracker.
* In this case the state will revert to the last transaction state record in the index.
*/
this.state = getTrackerState();
/*
* The fromCommitTime tells getSomeTransactions() where to start, this actually fairly straight forward.
*
* What makes this code so tricky to understand is the state.getTimeToStopIndexing().
*
* There are two scenarios to keep in mind:
*
* 1) Full re-index: In this scenario the state.getTimeToStopIndexing() will never stop the indexing.
*
* 2) Up-to-date indexing: This is where state.getTimeToStopIndexing() gets interesting. In this scenario
* the Solr index is already up to date with the repo and it is tracking new transactions. The state.getTimeToStopIndexing()
* in this scenario causes the getSomeTransactions() call to stop returning results if it finds a transaction
* beyond a specific point in time. This will break out of this loop and end the tracker run.
*
* The next time the metadata tracker runs the "continueState()" method applies the "hole retention"
* to state.getLastGoodTxCommitTimeInIndex(). This causes the state.getLastGoodTxCommitTimeInIndex() to scan
* for prior transactions that might have been missed.
*
*/
Long fromCommitTime = getTxFromCommitTime(txnsFound,
state.getLastIndexedTxCommitTime() == 0 ? state.getLastGoodTxCommitTimeInIndex() : state.getLastIndexedTxCommitTime());
log.debug("#### Check txnsFound : " + txnsFound.size());
log.debug("======= fromCommitTime: " + fromCommitTime);
// When using DB_ID_RANGE, fromCommitTime cannot be before the commit time of the first transaction
// for the DB_ID_RANGE to be indexed and commit time of the last transaction cannot be lower than fromCommitTime.
// When there isn't nodes in that range, -1 is returned as commit times
boolean shardOutOfRange = false;
// Get transaction list to be indexed
if (docRouter instanceof DBIDRangeRouter && txIntervalCommitTimeServiceAvailable)
{
DBIDRangeRouter dbIdRangeRouter = (DBIDRangeRouter) docRouter;
Pair<Long, Long> commitTimes = client.getTxIntervalCommitTime(coreName,
dbIdRangeRouter.getStartRange(), dbIdRangeRouter.getEndRange());
Long shardMinCommitTime = commitTimes.getFirst();
Long shardMaxCommitTime = commitTimes.getSecond();
// Node Range it's not still available in repository
if (shardMinCommitTime == -1)
{
log.debug(
"#### [DB_ID_RANGE] No nodes in range [{}-{}] "
+ "exist in the repository. Indexing only latest transaction.",
dbIdRangeRouter.getStartRange(), dbIdRangeRouter.getEndRange());
shardOutOfRange = true;
}
if (fromCommitTime > shardMaxCommitTime)
{
log.debug("#### [DB_ID_RANGE] Last commit time is greater that max commit time in in range [{}-{}]. "
+ "Indexing only latest transaction.",
dbIdRangeRouter.getStartRange(), dbIdRangeRouter.getEndRange());
shardOutOfRange = true;
}
// Initial commit time for Node Range is greater than calculated from commit time
if (fromCommitTime < shardMinCommitTime)
{
log.debug("#### [DB_ID_RANGE] Skipping transactions from {} to {}",
fromCommitTime, shardMinCommitTime);
fromCommitTime = shardMinCommitTime;
}
transactions = getDBIDRangeTransactions(fromCommitTime, txnsFound);
}
log.debug("#### Get txn from commit time: " + fromCommitTime);
transactions = getSomeTransactions(txnsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, 2000,
state.getTimeToStopIndexing());
// When transactions are out of Shard range, only the latest transaction needs to be indexed
// in order to preserve the state up-to-date of the MetadataTracker
if (shardOutOfRange)
else
{
Transaction latestTransaction = new Transaction();
latestTransaction.setCommitTimeMs(transactions.getMaxTxnCommitTime());
latestTransaction.setId(transactions.getMaxTxnId());
transactions = new Transactions(
Arrays.asList(latestTransaction),
transactions.getMaxTxnCommitTime(),
transactions.getMaxTxnId());
log.debug("#### [DB_ID_RANGE] Latest transaction to be indexed {}", latestTransaction);
transactions = getSomeTransactions(txnsFound, fromCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions,
state.getTimeToStopIndexing());
}
setLastTxCommitTimeAndTxIdInTrackerState(transactions, state);
log.debug("Scanning transactions ...");
if (transactions.getTransactions().size() > 0) {
log.info(".... from " + transactions.getTransactions().get(0));
log.info(".... to " + transactions.getTransactions().get(transactions.getTransactions().size() - 1));
} else {
log.info(".... none found after lastTxCommitTime "
+ ((txnsFound.size() > 0) ? txnsFound.getLast().getCommitTimeMs() : state
.getLastIndexedTxCommitTime()));
// Remove transactions already indexed
transactions = removeIndexedTransactions(transactions);
if (transactions.getTransactions().size() > 0)
{
LOGGER.info("{}-[CORE {}] Found {} transactions after lastTxCommitTime {}, transactions from {} to {}",
Thread.currentThread().getId(),
coreName,
transactions.getTransactions().size(),
fromCommitTime,
transactions.getTransactions().get(0),
transactions.getTransactions().get(transactions.getTransactions().size() - 1));
}
else
{
LOGGER.info("{}-[CORE {}] No transaction found after lastTxCommitTime {}",
Thread.currentThread().getId(),
coreName,
((txnsFound.size() > 0) ? txnsFound.getLast().getCommitTimeMs() : state.getLastIndexedTxCommitTime()));
}
ArrayList<Transaction> txBatch = new ArrayList<>();
// 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()) {
/*
* isInIndex is used to ensure transactions that are being re-pulled due to "hole retention" are not re-indexed if
* they have already been indexed.
*
* The logic in infoSrv.txnInIndex() first checks an in-memory LRUcache for the txnId. If it doesn't find it in the cache
* it checks the index. The LRUCache is only needed for txnId's that have been indexed but are not yet visible in the index for
* one of two reasons:
*
* 1) The commit tracker has not yet committed the transaction.
* 2) The txnId has been committed to the index but the new searcher has not yet been warmed.
*
* This means that to ensure txnId's are not needlessly reprocessed during hole retention, the LRUCache must be large
* enough to cover the time between when a txnId is indexed and when it becomes visible.
*/
boolean isInIndex = (infoSrv.txnInIndex(info.getId(), true) && info.getCommitTimeMs() <= state.getLastIndexedTxCommitTime());
if (isInIndex) {
txnsFound.add(info);
} else {
// 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) > this.transactionDocsBatchSize) {
docCount += indexBatchOfTransactions(txBatch);
totalUpdatedDocs += docCount;
for (Transaction scheduledTx : txBatch) {
txnsFound.add(scheduledTx);
txsIndexed.add(scheduledTx);
}
txBatch.clear();
}
if (LOGGER.isTraceEnabled())
{
LOGGER.trace("{}-[CORE {}] Tracking {} Transactions. Current Transaction Id to be indexed: {}",
Thread.currentThread().getId(), coreName, transactions.getTransactions().size(), info.getId());
}
if (docCount > batchCount) {
indexTransactionsAfterAsynchronous(txsIndexed, state);
long endElapsed = System.nanoTime();
trackerStats.addElapsedNodeTime(docCount, endElapsed - startElapsed);
startElapsed = endElapsed;
docCount = 0;
// 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;
}
//Release the write lock allowing the commit tracker to run.
this.getWriteLock().release();
//Re-acquire the write lock and keep indexing.
this.getWriteLock().acquire();
txBatch.add(info);
if (getUpdateAndDeleteCount(txBatch) > transactionDocsBatchSize) {
txBatches.add(txBatch);
txBatch = new ArrayList<>();
}
checkShutdown();
}
// Index any remaining transactions bringing the index to a consistent state so the CommitTracker can commit if need be.
if (!txBatch.isEmpty()) {
if (this.getUpdateAndDeleteCount(txBatch) > 0) {
docCount += indexBatchOfTransactions(txBatch);
totalUpdatedDocs += docCount;
}
for (Transaction scheduledTx : txBatch) {
txnsFound.add(scheduledTx);
txsIndexed.add(scheduledTx);
}
txBatch.clear();
if (!txBatch.isEmpty())
{
txBatches.add(txBatch);
}
// 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);
totalUpdatedDocs += docCount;
// Add the transactions as found to avoid processing them again in the next iteration
batch.forEach(transaction -> txnsFound.add(transaction));
if (txsIndexed.size() > 0) {
indexTransactionsAfterAsynchronous(txsIndexed, state);
// Index the transactions
indexTransactionsAfterWorker(batch);
long endElapsed = System.nanoTime();
trackerStats.addElapsedNodeTime(docCount, endElapsed - startElapsed);
startElapsed = endElapsed;
docCount = 0;
}
setLastTxCommitTimeAndTxIdInTrackerState(transactions);
}
catch(Exception e)
{
@@ -916,10 +1015,14 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
}
while ((transactions.getTransactions().size() > 0) && (upToDate == false));
log.debug("total number of docs with metadata updated: " + totalUpdatedDocs);
LOGGER.info("{}-[CORE {}] Tracked {} DOCs", Thread.currentThread().getId(), coreName, totalUpdatedDocs);
}
private void setLastTxCommitTimeAndTxIdInTrackerState(Transactions transactions, TrackerState state)
/**
* Update latest transaction indexed in MetadataTracker state
* @param transactions List of transactions indexed
*/
private void setLastTxCommitTimeAndTxIdInTrackerState(Transactions transactions)
{
Long maxTxnCommitTime = transactions.getMaxTxnCommitTime();
if (maxTxnCommitTime != null)
@@ -934,13 +1037,17 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
}
}
private void indexTransactionsAfterAsynchronous(HashSet<Transaction> txsIndexed, TrackerState state)
/**
* Index transactions and update state of the tracker
* @param txsIndexed List of transactions to be indexed
* @throws IOException
*/
private void indexTransactionsAfterWorker(List<Transaction> txsIndexed)
throws IOException
{
waitForAsynchronous();
for (Transaction tx : txsIndexed)
{
super.infoSrv.indexTransaction(tx, true);
infoSrv.indexTransaction(tx, true);
// Transactions are ordered by commit time and tie-broken by tx id
if (tx.getCommitTimeMs() > state.getLastIndexedTxCommitTime()
|| tx.getCommitTimeMs() == state.getLastIndexedTxCommitTime()
@@ -952,9 +1059,13 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
trackerStats.addTxDocs((int) (tx.getDeletes() + tx.getUpdates()));
}
txsIndexed.clear();
//super.infoSrv.commit();
}
/**
* 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;
@@ -965,12 +1076,25 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
return count;
}
private int indexBatchOfTransactions(List<Transaction> txBatch) throws AuthenticationException, IOException, JSONException
{
int nodeCount = 0;
/**
* Index a batch of transactions.
*
* Updated or deleted nodes from these transactions are also packed into batches in order to get
* 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 {
// Skip transactions without modifications (updates, deletes)
ArrayList<Transaction> nonEmptyTxs = new ArrayList<>(txBatch.size());
GetNodesParameters gnp = new GetNodesParameters();
ArrayList<Long> txIds = new ArrayList<Long>();
ArrayList<Long> txIds = new ArrayList<>();
for (Transaction tx : txBatch)
{
if (tx.getUpdates() > 0 || tx.getDeletes() > 0)
@@ -979,51 +1103,48 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
txIds.add(tx.getId());
}
}
// Get Nodes Id properties for every transaction
GetNodesParameters gnp = new GetNodesParameters();
gnp.setTransactionIds(txIds);
gnp.setStoreProtocol(storeRef.getProtocol());
gnp.setStoreIdentifier(storeRef.getIdentifier());
updateShardProperty();
shardProperty.ifPresent(p -> gnp.setShardProperty(p));
gnp.setCoreName(coreName);
List<Node> nodes = client.getNodes(gnp, Integer.MAX_VALUE);
ArrayList<Node> nodeBatch = new ArrayList<>();
for (Node node : nodes)
if (LOGGER.isDebugEnabled())
{
if (log.isDebugEnabled())
{
log.debug(node.toString());
}
nodeBatch.add(node);
if (nodeBatch.size() > nodeBatchSize)
{
nodeCount += nodeBatch.size();
NodeIndexWorkerRunnable niwr = new NodeIndexWorkerRunnable(this.threadHandler, nodeBatch, this.infoSrv);
this.threadHandler.scheduleTask(niwr);
nodeBatch = new ArrayList<>();
}
LOGGER.debug("{}-[CORE {}] Found {} Nodes to be indexed from Transactions: {}", Thread.currentThread().getId(),
coreName, nodes.size(), txIds);
}
if (nodeBatch.size() > 0)
{
nodeCount += nodeBatch.size();
NodeIndexWorkerRunnable niwr = new NodeIndexWorkerRunnable(this.threadHandler, nodeBatch, this.infoSrv);
this.threadHandler.scheduleTask(niwr);
nodeBatch = new ArrayList<>();
}
return nodeCount;
// Group the nodes in batches of nodeBatchSize (or less)
List<List<Node>> nodeBatches = Lists.partition(nodes, nodeBatchSize);
Integer processedNodes = forkJoinPool.submit(() ->
nodeBatches.parallelStream().map(batch -> {
new NodeIndexWorker(batch, infoSrv).run();
return batch.size();
}).reduce(0, Integer::sum)).get();
return processedNodes;
}
class NodeIndexWorkerRunnable extends AbstractWorkerRunnable
/**
* Node Indexing class running synchronously all the tracking operations.
*/
class NodeIndexWorker extends AbstractWorker
{
InformationServer infoServer;
List<Node> nodes;
NodeIndexWorkerRunnable(QueueHandler queueHandler, List<Node> nodes, InformationServer infoServer)
NodeIndexWorker(List<Node> nodes, InformationServer infoServer)
{
super(queueHandler);
this.infoServer = infoServer;
this.nodes = nodes;
}
@@ -1186,7 +1307,7 @@ public class MetadataTracker extends CoreStatePublisher implements Tracker
long endTime = System.currentTimeMillis() + infoSrv.getHoleRetention();
DO: do
{
transactions = getSomeTransactions(txnsFound, lastTxCommitTime, TIME_STEP_1_HR_IN_MS, 2000, endTime);
transactions = getSomeTransactions(txnsFound, lastTxCommitTime, TIME_STEP_1_HR_IN_MS, maxNumberOfTransactions, endTime);
for (Transaction info : transactions.getTransactions())
{
// include
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java
+19
View File
@@ -30,6 +30,8 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.alfresco.error.AlfrescoRuntimeException;
@@ -102,6 +104,20 @@ public class ModelTracker extends AbstractTracker implements Tracker
private ReentrantReadWriteLock modelLock = new ReentrantReadWriteLock();
private volatile boolean hasModels = false;
private File alfrescoModelDir;
// Share run and write locks across all ModelTracker threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
public ModelTracker(String solrHome, Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
@@ -116,6 +132,9 @@ public class ModelTracker extends AbstractTracker implements Tracker
}
loadPersistedModels();
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
@Override
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/QueueHandler.java
-24
View File
@@ -1,24 +0,0 @@
/*
* Copyright (C) 2005-2014 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* 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/>.
*/
package org.alfresco.solr.tracker;
public interface QueueHandler
{
void removeFromQueueAndProdHead(AbstractWorkerRunnable job);
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SlaveCoreStatePublisher.java
+21
View File
@@ -30,7 +30,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
/**
* Despite belonging to the Tracker ecosystem, this component is actually a publisher, which periodically informs
@@ -51,6 +54,21 @@ public class SlaveCoreStatePublisher extends CoreStatePublisher
{
private static final Logger LOGGER = LoggerFactory.getLogger(SlaveCoreStatePublisher.class);
// Share run and write locks across all SlaveCoreStatePublisher threads
private static Map<String, Semaphore> RUN_LOCK_BY_CORE = new ConcurrentHashMap<>();
private static Map<String, Semaphore> WRITE_LOCK_BY_CORE = new ConcurrentHashMap<>();
@Override
public Semaphore getWriteLock()
{
return WRITE_LOCK_BY_CORE.get(coreName);
}
@Override
public Semaphore getRunLock()
{
return RUN_LOCK_BY_CORE.get(coreName);
}
public SlaveCoreStatePublisher(
boolean isMaster,
Properties coreProperties,
@@ -59,6 +77,9 @@ public class SlaveCoreStatePublisher extends CoreStatePublisher
SolrInformationServer informationServer)
{
super(isMaster, coreProperties, repositoryClient, name, informationServer, NODE_STATE_PUBLISHER);
RUN_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
WRITE_LOCK_BY_CORE.put(coreName, new Semaphore(1, true));
}
@Override
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ThreadHandler.java
-104
View File
@@ -1,104 +0,0 @@
/*
* Copyright (C) 2005-2014 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* 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/>.
*/
package org.alfresco.solr.tracker;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.alfresco.solr.tracker.pool.DefaultTrackerPoolFactory;
import org.alfresco.solr.tracker.pool.TrackerPoolFactory;
/**
* This class handles threads for trackers.
*
* @author Ahmed Owian
*/
public class ThreadHandler implements QueueHandler
{
/** the instance that will be given out by the factory */
private ThreadPoolExecutor threadPool;
private LinkedBlockingQueue<AbstractWorkerRunnable> threadQueue = new LinkedBlockingQueue<>();
private ReentrantReadWriteLock threadLock = new ReentrantReadWriteLock(true);
public ThreadHandler(Properties p, String coreName, String trackerName)
{
// construct the instance
TrackerPoolFactory trackerPoolFactory = new DefaultTrackerPoolFactory(p, coreName, trackerName);
threadPool = trackerPoolFactory.create();
}
public void scheduleTask(AbstractWorkerRunnable awr)
{
try
{
threadLock.writeLock().lock();
// Add the runnable to the queue to ensure ordering
threadQueue.add(awr);
}
finally
{
threadLock.writeLock().unlock();
}
threadPool.execute(awr);
}
/**
* Removes the job from the queue and notifies the HEAD
*/
public void removeFromQueueAndProdHead(AbstractWorkerRunnable job)
{
try
{
threadLock.writeLock().lock();
// Remove self from head of queue
threadQueue.remove(job);
}
finally
{
threadLock.writeLock().unlock();
}
}
/**
* Read-safe method to peek at the head of the queue
*/
public AbstractWorkerRunnable peekHeadReindexWorker()
{
try
{
threadLock.readLock().lock();
return threadQueue.peek();
}
finally
{
threadLock.readLock().unlock();
}
}
public void shutDownThreadPool()
{
if (threadPool != null)
{
threadPool.shutdownNow();
}
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/TrackerJob.java
+5
View File
@@ -18,6 +18,7 @@
*/
package org.alfresco.solr.tracker;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
@@ -28,8 +29,12 @@ import org.slf4j.LoggerFactory;
* Generic Solr tracker job, allowing Quartz to initiate an index update from
* a {@link Tracker} regardless of specific implementation.
*
* Concurrent execution is disallowed, as no parallel work can be done when
* indexing contents from the repository.
*
* @author Matt Ward
*/
@DisallowConcurrentExecution
public class TrackerJob implements Job
{
public static final String JOBDATA_TRACKER_KEY = "TRACKER";
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/pool/DefaultTrackerPoolFactory.java
+1 -1
View File
@@ -99,7 +99,7 @@ public class DefaultTrackerPoolFactory implements TrackerPoolFactory
{
case "AclTracker":
corePoolSize = parseConfig("alfresco.acl.tracker.corePoolSize", p, corePoolSize);
maximumPoolSize = parseConfig("alfresco.acl.tracker.maximumPoolSize", p, maximumPoolSize);
maximumPoolSize = parseConfig("alfresco.acl.tracker.maximumPoolSize", p, maximumPoolSize);
keepAliveTime = parseConfig("alfresco.acl.tracker.keepAliveTime", p, keepAliveTime);
threadPriority = parseConfig("alfresco.acl.tracker.threadPriority", p, threadPriority);
threadDaemon = parseConfigBoolean("alfresco.acl.tracker.threadDaemon", p);
search-services/alfresco-search/src/main/resources/solr/instance/templates/rerank/conf/solrcore.properties
+11 -1
View File
@@ -122,12 +122,22 @@ solr.maxBooleanClauses=10000
# Batch fetch
alfresco.transactionDocsBatchSize=500
#Max number of transactions fetched by metadata tracker
#alfresco.maxNumberOfTransactions=
alfresco.transactionDocsBatchSize=2000
alfresco.nodeBatchSize=100
alfresco.changeSetAclsBatchSize=500
alfresco.aclBatchSize=100
alfresco.contentReadBatchSize=100
alfresco.contentUpdateBatchSize=1000
alfresco.cascadeNodeBatchSize=10
# Trackers thread pools
#alfresco.metadataTrackerMaxParallelism=
#alfresco.aclTrackerMaxParallelism=
#alfresco.contentTrackerMaxParallelism=
#alfresco.cascadeTrackerMaxParallelism
# Warming
search-services/alfresco-search/src/test/java/org/alfresco/solr/AbstractAlfrescoDistributedIT.java
+2 -1
View File
@@ -850,7 +850,8 @@ public abstract class AbstractAlfrescoDistributedIT extends SolrITInitializer
public static void indexTransaction(Transaction transaction, List<Node> nodes, List<NodeMetaData> nodeMetaDatas)
{
//First map the nodes to a transaction.
// First map the nodes to a transaction.
nodes.stream().forEach(node -> node.setTxnId(transaction.getId()));
SOLRAPIQueueClient.nodeMap.put(transaction.getId(), nodes);
//Next map a node to the NodeMetaData
search-services/alfresco-search/src/test/java/org/alfresco/solr/SolrITInitializer.java
+2 -1
View File
@@ -1,6 +1,7 @@
package org.alfresco.solr;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
import org.alfresco.solr.basics.RandomSupplier;
import org.alfresco.solr.client.SOLRAPIQueueClient;
import org.apache.commons.io.FileUtils;
@@ -67,7 +68,7 @@ import static org.alfresco.solr.AlfrescoSolrUtils.createCoreUsingTemplate;
* @since solr 1.5
* @author Michael Suzuki
*/
@ThreadLeakLingering(linger = 5000)
@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
public abstract class SolrITInitializer extends SolrTestCaseJ4
{
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
search-services/alfresco-search/src/test/java/org/alfresco/solr/basics/SolrResponsesComparator.java
+40 -21
View File
@@ -1,5 +1,6 @@
package org.alfresco.solr.basics;
import com.google.common.collect.Sets;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocument;
@@ -12,8 +13,10 @@ import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class SolrResponsesComparator
{
@@ -51,6 +54,7 @@ public class SolrResponsesComparator
handle.put("_version_", SKIP);
handle.put("_original_parameters_", SKIP);
handle.put("spellcheck-extras", SKIP); // No longer used can be removed in Solr 6.
handle.put("FIELDS", UNORDERED);
}
@@ -85,27 +89,6 @@ public class SolrResponsesComparator
public void compareResponses(QueryResponse a, QueryResponse b)
{
if (System.getProperty("remove.version.field") != null)
{
// we don't care if one has a version and the other doesnt -
// control vs distrib
// TODO: this should prob be done by adding an ignore on _version_
// rather than mutating the responses?
if (a.getResults() != null)
{
for (SolrDocument doc : a.getResults())
{
doc.removeFields("_version_");
}
}
if (b.getResults() != null)
{
for (SolrDocument doc : b.getResults())
{
doc.removeFields("_version_");
}
}
}
compareSolrResponses(a, b);
}
@@ -233,6 +216,29 @@ public class SolrResponsesComparator
return compare1(b, a, flags, handle);
}
public static String compare(Set a, Set b, int flags, Map<String, Integer> handle)
{
String cmp;
cmp = compare1(a, b, flags, handle);
if (cmp != null)
return cmp;
return compare1(b, a, flags, handle);
}
private static String compare1(Set a, Set b, int flags, Map<String, Integer> handle) {
for (Object valA : a)
{
int flagsa = flags(handle, valA);
if ((flagsa & SKIP) != 0)
continue;
if (!b.contains(valA))
{
return "[" + valA + "]==null";
}
}
return null;
}
public static String compare(SolrDocument a, SolrDocument b, int flags, Map<String, Integer> handle)
{
return compare(a.getFieldValuesMap(), b.getFieldValuesMap(), flags, handle);
@@ -315,10 +321,22 @@ public class SolrResponsesComparator
public static String compare(Object[] a, Object[] b, int flags, Map<String, Integer> handle)
{
boolean ordered = (flags & UNORDERED) == 0;
if (a.length != b.length)
{
return ".length:" + a.length + "!=" + b.length;
}
if (!ordered)
{
Set<Object> setA = Sets.newHashSet(a);
Set<Object> setB = Sets.newHashSet(b);
return compare(setA, setB, flags, handle);
}
for (int i = 0; i < a.length; i++)
{
String cmp = compare(a[i], b[i], flags, handle);
@@ -328,6 +346,7 @@ public class SolrResponsesComparator
return null;
}
public static String compare(Object a, Object b, int flags, Map<String, Integer> handle)
{
if (a == b)
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerIT.java
+4 -6
View File
@@ -53,14 +53,12 @@ public class ContentTrackerIT
private TrackerStats trackerStats;
private int UPDATE_BATCH = 2;
private int READ_BATCH = 400;
@Before
public void setUp() throws Exception
{
doReturn("workspace://SpacesStore").when(props).getProperty(eq("alfresco.stores"), anyString());
doReturn("" + UPDATE_BATCH).when(props).getProperty(eq("alfresco.contentUpdateBatchSize"), anyString());
doReturn("" + READ_BATCH).when(props).getProperty(eq("alfresco.contentReadBatchSize"), anyString());
when(srv.getTrackerStats()).thenReturn(trackerStats);
this.contentTracker = new ContentTracker(props, repositoryClient, coreName, srv);
@@ -103,14 +101,14 @@ public class ContentTrackerIT
doc.tenant = "2";
docs2.add(doc);
}
when(this.srv.getDocsWithUncleanContent(anyInt(), anyInt()))
when(this.srv.getDocsWithUncleanContent())
.thenReturn(docs1)
.thenReturn(docs2)
.thenReturn(emptyList);
this.contentTracker.doTrack("anIterationId");
InOrder order = inOrder(srv);
order.verify(srv).getDocsWithUncleanContent(0, READ_BATCH);
order.verify(srv).getDocsWithUncleanContent();
/*
* I had to make each bunch of calls have different parameters to prevent Mockito from incorrectly failing
@@ -126,13 +124,13 @@ public class ContentTrackerIT
order.verify(srv).updateContentToIndexAndCache(thirdDoc.dbId, thirdDoc.tenant);
order.verify(srv).commit();
order.verify(srv).getDocsWithUncleanContent(0 + READ_BATCH, READ_BATCH);
order.verify(srv).getDocsWithUncleanContent();
// From docs2
order.verify(srv, times(UPDATE_BATCH)).updateContentToIndexAndCache(2l, "2");
order.verify(srv).commit();
order.verify(srv).getDocsWithUncleanContent(0 + READ_BATCH + READ_BATCH, READ_BATCH);
order.verify(srv).getDocsWithUncleanContent();
}
@Test
public void typeCheck()
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedExpandDbidRangeAlfrescoSolrTrackerIT.java
+13
View File
@@ -18,6 +18,7 @@
*/
package org.alfresco.solr.tracker;
import org.alfresco.repo.search.adaptor.lucene.QueryConstants;
import org.alfresco.solr.AbstractAlfrescoDistributedIT;
import org.alfresco.solr.SolrInformationServer;
import org.alfresco.solr.client.Acl;
@@ -27,6 +28,9 @@ 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.lucene.util.LuceneTestCase;
import org.apache.solr.SolrTestCaseJ4;
@@ -42,6 +46,7 @@ import java.util.Properties;
import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DOC_TYPE;
import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_SOLR4_ID;
import static org.alfresco.solr.AlfrescoSolrUtils.MAX_WAIT_TIME;
import static org.alfresco.solr.AlfrescoSolrUtils.getAcl;
import static org.alfresco.solr.AlfrescoSolrUtils.getAclChangeSet;
import static org.alfresco.solr.AlfrescoSolrUtils.getAclReaders;
@@ -95,6 +100,14 @@ public class DistributedExpandDbidRangeAlfrescoSolrTrackerIT extends AbstractAlf
bulkAcls,
bulkAclReaders);
//Check for the ACL state stamp.
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,
bulkAclChangeSet.getId(), bulkAclChangeSet.getId() + 1, true, false), BooleanClause.Occur.MUST));
BooleanQuery waitForQuery = builder.build();
waitForDocCount(waitForQuery, 1, MAX_WAIT_TIME);
SolrQueryResponse response0 = rangeCheck(0);
NamedList values0 = response0.getValues();
//{start=0,end=100,nodeCount=0,maxDbid=0,density=NaN,expand=0,expanded=false}
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/pool/DefaultTrackerPoolFactoryTest.java
+7 -7
View File
@@ -82,7 +82,7 @@ public class DefaultTrackerPoolFactoryTest
public void testAclDefaultProperties()
{
poolFactory = new DefaultTrackerPoolFactory(properties, "TheCore", "AclTracker");
tpe = poolFactory.create();
assertEquals(4, tpe.getCorePoolSize());
assertEquals(4, tpe.getMaximumPoolSize());
@@ -96,7 +96,7 @@ public class DefaultTrackerPoolFactoryTest
properties.put("alfresco.acl.tracker.keepAliveTime", "200");
poolFactory = new DefaultTrackerPoolFactory(properties, "TheCore", "AclTracker");
tpe = poolFactory.create();
assertEquals(30, tpe.getCorePoolSize());
assertEquals(40, tpe.getMaximumPoolSize());
assertEquals(200, tpe.getKeepAliveTime(TimeUnit.SECONDS));
@@ -120,7 +120,7 @@ public class DefaultTrackerPoolFactoryTest
properties.put("alfresco.content.tracker.keepAliveTime", "201");
poolFactory = new DefaultTrackerPoolFactory(properties, "TheCore", "ContentTracker");
tpe = poolFactory.create();
assertEquals(100, tpe.getCorePoolSize());
assertEquals(140, tpe.getMaximumPoolSize());
assertEquals(201, tpe.getKeepAliveTime(TimeUnit.SECONDS));
@@ -129,9 +129,9 @@ public class DefaultTrackerPoolFactoryTest
public void testMetaDataDefaultProperties()
{
poolFactory = new DefaultTrackerPoolFactory(properties, "TheCore", "MetadataTracker");
tpe = poolFactory.create();
assertEquals(4, tpe.getCorePoolSize());
assertEquals(4, tpe.getMaximumPoolSize());
assertEquals(120, tpe.getKeepAliveTime(TimeUnit.SECONDS));
@@ -143,9 +143,9 @@ public class DefaultTrackerPoolFactoryTest
properties.put("alfresco.metadata.tracker.maximumPoolSize", "140");
properties.put("alfresco.metadata.tracker.keepAliveTime", "201");
poolFactory = new DefaultTrackerPoolFactory(properties, "TheCore", "MetadataTracker");
tpe = poolFactory.create();
assertEquals(100, tpe.getCorePoolSize());
assertEquals(140, tpe.getMaximumPoolSize());
assertEquals(201, tpe.getKeepAliveTime(TimeUnit.SECONDS));
search-services/alfresco-search/src/test/java/org/alfresco/solr/transformer/CachedDocTransformerIT.java
+8 -8
View File
@@ -18,6 +18,7 @@
*/
package org.alfresco.solr.transformer;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.search.adaptor.lucene.QueryConstants;
import org.alfresco.service.cmr.repository.datatype.DefaultTypeConverter;
@@ -34,7 +35,6 @@ 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.lucene.util.LuceneTestCase;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocument;
@@ -56,7 +56,7 @@ import static org.alfresco.solr.AlfrescoSolrUtils.getTransaction;
import static org.alfresco.solr.AlfrescoSolrUtils.indexAclChangeSet;
import static org.alfresco.solr.AlfrescoSolrUtils.list;
@LuceneTestCase.SuppressCodecs({"Appending","Lucene3x","Lucene40","Lucene41","Lucene42","Lucene43", "Lucene44", "Lucene45","Lucene46","Lucene47","Lucene48","Lucene49"})
@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
@SolrTestCaseJ4.SuppressSSL
public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
{
@@ -80,7 +80,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
{
putHandleDefaults();
//Test 1: Running a simple query without invoking CachedDocTransformer, expected to see id,DBID and _version_
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "sort", "id asc", "shards.qt", "/afts"));
assertNotNull(resp);
SolrDocumentList results = resp.getResults();
assertEquals("Expecting 5 rows",5, results.size());
@@ -106,7 +106,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
putHandleDefaults();
//Test 2: Running simple query with CachedDocTransformer, expected to see all fields returned
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","fl","*,[cached]"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts", "sort", "id asc", "fl", "*,[cached]"));
SolrDocument docWithAllFields = resp.getResults().get(0);
assertTrue(docWithAllFields.size() > 3);
@@ -128,7 +128,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
putHandleDefaults();
//Test 3: Running simple query with CachedDocTransformer, expected to see selected fields returned
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","fl","id,DBID,[cached]"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts", "sort", "id asc", "fl","id,DBID,[cached]"));
assertNotNull(resp);
SolrDocument docWithRequestedFields = resp.getResults().get(0);
@@ -143,7 +143,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
putHandleDefaults();
//Test 4: Running simple query with CachedDocTransformer on non default fields, expected to see selected fields returned
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","fl","id, cm_title,[cached]"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","sort", "id asc", "fl","id, cm_title,[cached]"));
assertNotNull(resp);
SolrDocument docWithRequestedFields3 = resp.getResults().get(0);
@@ -158,7 +158,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
{
putHandleDefaults();
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","fl","cm_name, score, [cached]"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts", "sort", "id asc", "fl","cm_name, score, [cached]"));
assertNotNull(resp);
SolrDocumentList results = resp.getResults();
SolrDocument docWithAllFields = results.get(0);
@@ -173,7 +173,7 @@ public class CachedDocTransformerIT extends AbstractAlfrescoDistributedIT
{
putHandleDefaults();
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","fl","cm_title, cm_created, DBID, score, [cached]"));
QueryResponse resp = query(getDefaultTestClient(), true, ALFRESCO_JSON, params("q", "*", "qt", "/afts", "shards.qt", "/afts","sort", "id asc", "fl","cm_title, cm_created, DBID, score, [cached]"));
assertNotNull(resp);
SolrDocumentList results = resp.getResults();
SolrDocument docWithAllFields = results.get(0);
search-services/alfresco-solrclient-lib/src/main/java/org/alfresco/solr/client/AclChangeSets.java
+79 -66
View File
@@ -1,66 +1,79 @@
/*
* #%L
* Alfresco Solr Client
* %%
* Copyright (C) 2005 - 2016 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.client;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
/**
* @author Andy
*
*/
public class AclChangeSets
{
private List<AclChangeSet> aclChangeSets;
private Long maxChangeSetCommitTime;
private Long maxChangeSetId;
AclChangeSets(List<AclChangeSet> aclChangeSets, Long maxChangeSetCommitTime, Long maxChangeSetId)
{
this.aclChangeSets = (aclChangeSets == null ? null : new ArrayList<>(aclChangeSets));
this.maxChangeSetCommitTime = maxChangeSetCommitTime;
this.maxChangeSetId = maxChangeSetId;
}
public List<AclChangeSet> getAclChangeSets()
{
return Collections.unmodifiableList(aclChangeSets);
}
public Long getMaxChangeSetCommitTime()
{
return maxChangeSetCommitTime;
}
public Long getMaxChangeSetId()
{
return maxChangeSetId;
}
}
/*
* #%L
* Alfresco Solr Client
* %%
* Copyright (C) 2005 - 2016 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.client;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
/**
* @author Andy
*
*/
public class AclChangeSets
{
private List<AclChangeSet> aclChangeSets;
private Long maxChangeSetCommitTime;
private Long maxChangeSetId;
AclChangeSets(List<AclChangeSet> aclChangeSets, Long maxChangeSetCommitTime, Long maxChangeSetId)
{
this.aclChangeSets = (aclChangeSets == null ? null : new ArrayList<>(aclChangeSets));
this.maxChangeSetCommitTime = maxChangeSetCommitTime;
this.maxChangeSetId = maxChangeSetId;
}
public AclChangeSets(List<AclChangeSet> aclChangeSets)
{
this.aclChangeSets = aclChangeSets;
if (!aclChangeSets.isEmpty())
{
this.maxChangeSetCommitTime = aclChangeSets.stream()
.max(Comparator.comparing(AclChangeSet::getCommitTimeMs)).get().getCommitTimeMs();
this.maxChangeSetId = aclChangeSets.stream()
.max(Comparator.comparing(AclChangeSet::getId)).get().getId();
}
}
public List<AclChangeSet> getAclChangeSets()
{
return Collections.unmodifiableList(aclChangeSets);
}
public Long getMaxChangeSetCommitTime()
{
return maxChangeSetCommitTime;
}
public Long getMaxChangeSetId()
{
return maxChangeSetId;
}
}
search-services/alfresco-solrclient-lib/src/main/java/org/alfresco/solr/client/SOLRAPIQueueClient.java
+1 -1
View File
@@ -246,7 +246,7 @@ public class SOLRAPIQueueClient extends SOLRAPIClient
{
//We have moved beyond this transaction.
}
else if(txn.getCommitTimeMs() > toCommitTime)
else if (toCommitTime != null && txn.getCommitTimeMs() > toCommitTime)
{
//We have not yet reached this transaction so break out of the loop
break;
search-services/alfresco-solrclient-lib/src/main/java/org/alfresco/solr/client/Transactions.java
+14 -1
View File
@@ -25,6 +25,7 @@
*/
package org.alfresco.solr.client;
import java.util.Comparator;
import java.util.List;
/**
@@ -44,7 +45,19 @@ public class Transactions
this.transactions = transactions;
this.maxTxnCommitTime = maxTxnCommitTime;
this.maxTxnId = maxTxnId;
}
}
public Transactions(List<Transaction> transactions)
{
this.transactions = transactions;
if (!transactions.isEmpty())
{
this.maxTxnCommitTime = transactions.stream()
.max(Comparator.comparing(Transaction::getCommitTimeMs)).get().getCommitTimeMs();
this.maxTxnId = transactions.stream()
.max(Comparator.comparing(Transaction::getId)).get().getId();
}
}
public List<Transaction> getTransactions()
{