Merge branch 'develop' into stable (9ee890b7) · Commits · Inteligr8 / asie-platform-module

shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java

+4 −4

Original line number	Diff line number	Diff line
		@@ -44,10 +44,10 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript {
		public void reconciled(long nodeDbId) {
		if (includeReconciled) {
		@SuppressWarnings("unchecked")
		List<Long> unreconciledNodeDbIds = (List<Long>) responseMap.get("reconciled");
		if (unreconciledNodeDbIds == null)
		responseMap.put("reconciled", unreconciledNodeDbIds = new LinkedList<>());
		unreconciledNodeDbIds.add(nodeDbId);
		List<Long> reconciledNodeDbIds = (List<Long>) responseMap.get("reconciled");
		if (reconciledNodeDbIds == null)
		responseMap.put("reconciled", reconciledNodeDbIds = new LinkedList<>());
		reconciledNodeDbIds.add(nodeDbId);
		}
		}

shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java

+0 −4

Original line number	Diff line number	Diff line
		@@ -233,10 +233,6 @@ public abstract class AbstractNodeActionService implements DisposableBean {
		if (this.shardRegistry == null)
		throw new UnsupportedOperationException("ACS instances without a sharding configuration are not yet implemented");

		// SearchParameters searchParams = new SearchParameters();
		// searchParams.setLanguage(SearchService.LANGUAGE_FTS_ALFRESCO);
		// searchParams.setQuery("@" + this.formatForFts(ContentModel.PROP_NODE_DBID) + ":" + nodeDbId);

		List<com.inteligr8.alfresco.asie.model.ShardInstance> instances = new LinkedList<>();

		// we need a ShardRegistry method like getIndexSlice, but

shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java

+79 −28

Original line number	Diff line number	Diff line
		@@ -75,6 +75,12 @@ public class AcsReconcileService implements DisposableBean {
		@Value("${inteligr8.asie.reconciliation.concurrency}")
		private int concurrency;

		@Value("${inteligr8.asie.reconciliation.waitAfterSolrNodeActionMillis}")
		private long waitAfterSolrNodeActionMillis;

		@Value("${inteligr8.asie.reconciliation.waitAfterSolrNodeReconcileMillis}")
		private long waitAfterSolrNodeReconcileMillis;

		@Override
		public void destroy() {
		ExecutorService executor = this.executorManager.get("solr-reconcile");
		@@ -100,7 +106,7 @@ public class AcsReconcileService implements DisposableBean {
		*
		* There are two sets of parameters regarding timeouts. The queue timeouts
		* are for how long the requesting thread should wait for a full queue to
		* open up space for new re-index executions. The execution timeouts are
		* open up space for new reconcile executions. The execution timeouts are
		* for how long the execution should be allowed to take once dequeued.
		* There is no timeout for how long the execution is queued.
		*
		@@ -111,10 +117,10 @@ public class AcsReconcileService implements DisposableBean {
		* @param callback A callback to process multiple returned values from the re-index.
		* @param queueTimeout A timeout for how long the calling thread should wait for space on the queue.
		* @param queueUnit The time units for the `queueTimeout`.
		* @param execTimeout A timeout for the elapsed time the reindex execution should take when dequeued.
		* @param execTimeout A timeout for the elapsed time the reconcile execution should take when dequeued.
		* @param execUnit The time units for the `execTimeout`.
		* @throws TimeoutException Either the queue or execution timeout lapsed.
		* @throws InterruptedException The re-index was interrupted (server shutdown).
		* @throws InterruptedException The reconciliation was interrupted (server shutdown).
		*/
		public void reconcile(
		long fromDbId, long toDbId, Integer nodesChunkSize,
		@@ -123,18 +129,10 @@ public class AcsReconcileService implements DisposableBean {
		ReconcileCallback callback,
		long queueTimeout, TimeUnit queueUnit,
		long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
		if (nodesChunkSize == null)
		nodesChunkSize = this.nodesChunkSize;
		if (this.logger.isTraceEnabled())
		this.logger.trace("reconcile({}, {}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));

		CompositeFuture<Void> future = new CompositeFuture<>();

		for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {
		long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
		future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));
		future.purge(true);
		}
		Future<Void> future = this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit);

		try {
		future.get(execTimeout, execUnit);
		@@ -143,26 +141,51 @@ public class AcsReconcileService implements DisposableBean {
		}
		}

		/**
		* This method reconciles the specified node range between ACS and Solr.
		* The node range is specified using the ACS unique database identifiers.
		* There is no other reasonably efficient attack vector. The callback
		* handles all the return values. This is the synchronous alternative to
		* the other `reconcile` method.
		*
		* @param fromDbId A node database ID, inclusive.
		* @param toDbId A node database ID, exclusive.
		* @param indexUnreconciled For nodes not found in Solr, attempt to index against all applicable Solr instances.
		* @param reindexReconciled For nodes found in Solr, attempt to re-index against all applicable Solr instances.
		* @param callback A callback to process multiple returned values from the re-index.
		* @throws InterruptedException The reconciliation was interrupted (server shutdown).
		*/
		public Future<Void> reconcile(
		long fromDbId, long toDbId, Integer nodesChunkSize,
		boolean indexUnreconciled,
		boolean reindexReconciled,
		ReconcileCallback callback) throws InterruptedException {
		this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);

		try {
		return this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null);
		} catch (TimeoutException te) {
		throw new RuntimeException("This should never happen: " + te.getMessage(), te);
		}
		}

		protected Future<Void> _reconcile(
		long fromDbId, long toDbId, Integer nodesChunkSize,
		boolean indexUnreconciled,
		boolean reindexReconciled,
		ReconcileCallback callback,
		long queueTimeout, TimeUnit queueUnit,
		long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
		if (nodesChunkSize == null)
		nodesChunkSize = this.nodesChunkSize;
		this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);

		CompositeFuture<Void> future = new CompositeFuture<>();

		try {
		for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {
		long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
		future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null));
		future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));
		future.purge(true);
		}
		} catch (TimeoutException te) {
		throw new RuntimeException("This should never happen: " + te.getMessage(), te);
		}

		return future;
		}
		@@ -211,11 +234,17 @@ public class AcsReconcileService implements DisposableBean {
		CompositeFuture<Void> future = new CompositeFuture<>();
		ThrottledThreadPoolExecutor executor = this.getExecutor();

		ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(
		"solr-reconcile",
		this.concurrency, this.concurrency, this.concurrentQueueSize,
		1L, TimeUnit.MINUTES);

		for (long _nodeDbId = fromDbId; _nodeDbId < toDbId; _nodeDbId++) {
		final long nodeDbId = _nodeDbId;
		this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId);

		Callable<Void> callable;
		boolean callingSolr = false;

		final int dbIdIndex = (int) (nodeDbId - fromDbId);
		if (nodeRefs[dbIdIndex] != null) {
		@@ -235,6 +264,8 @@ public class AcsReconcileService implements DisposableBean {
		return null;
		}
		};
		if (reindexReconciled)
		callingSolr = true;
		} else {
		callable = new Callable<Void>() {
		@Override
		@@ -243,6 +274,7 @@ public class AcsReconcileService implements DisposableBean {
		return null;
		}
		};
		callingSolr = true;
		}

		if (queueTimeout < 0L) {
		@@ -250,36 +282,42 @@ public class AcsReconcileService implements DisposableBean {
		} else {
		future.combine(executor.submit(callable, queueTimeout, queueUnit));
		}

		if (callingSolr && this.waitAfterSolrNodeReconcileMillis > 0L) {
		this.logger.trace("Waiting between each node reconcile");
		Thread.sleep(this.waitAfterSolrNodeReconcileMillis);
		}
		}

		return future;
		}

		public void reconcile(long nodeDbId,
		public boolean reconcile(long nodeDbId,
		boolean index,
		ReconcileCallback callback) throws InterruptedException, TimeoutException {
		NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
		if (nodeRef == null) {
		this.logger.trace("No such ACS node: {}; skipping ...", nodeDbId);
		return;
		return false;
		}

		if (!StoreRef.STORE_REF_WORKSPACE_SPACESSTORE.equals(nodeRef.getStoreRef())) {
		this.logger.trace("A deliberately ignored store in the DB is not indexed in Solr: {}: {}", nodeDbId, nodeRef);
		return;
		return false;
		}

		Set<QName> aspects = this.nodeService.getAspects(nodeRef);
		aspects.retainAll(this.ignoreNodesWithAspects);
		if (!aspects.isEmpty()) {
		this.logger.trace("A deliberately ignored node in the DB is not indexed in Solr: {}: {}: {}", nodeDbId, nodeRef, aspects);
		return;
		return false;
		}

		if (!index) {
		this.logger.debug("A node in the DB is not indexed in Solr: {}: {}", nodeDbId, nodeRef);
		this.reconcileLogger.info("UNRECONCILED: {} <=> {}", nodeDbId, nodeRef);
		callback.unreconciled(nodeDbId);
		return false;
		} else {
		this.logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);
		this.index(nodeDbId, nodeRef, callback);
		@@ -287,6 +325,7 @@ public class AcsReconcileService implements DisposableBean {
		// its results will be logged
		// the reconcile thread will continue independently
		// the callback will lag
		return true;
		}
		}

		@@ -321,7 +360,13 @@ public class AcsReconcileService implements DisposableBean {
		}
		};

		return this.indexService.index(nodeDbId, indexCallback);
		Future<Void> future = this.indexService.index(nodeDbId, indexCallback);

		if (this.waitAfterSolrNodeActionMillis > 0L) {
		Thread.sleep(this.waitAfterSolrNodeActionMillis);
		}

		return future;
		}

		public Future<Void> reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException {
		@@ -355,7 +400,13 @@ public class AcsReconcileService implements DisposableBean {
		}
		};

		return this.reindexService.reindex(nodeDbId, reindexCallback);
		Future<Void> future = this.reindexService.reindex(nodeDbId, reindexCallback);

		if (this.waitAfterSolrNodeActionMillis > 0L) {
		Thread.sleep(this.waitAfterSolrNodeActionMillis);
		}

		return future;
		}

		private String formatForFts(QName qname) {

shared/src/main/java/com/inteligr8/alfresco/asie/util/CompositeFuture.java

+16 −4

Original line number	Diff line number	Diff line
		@@ -79,12 +79,14 @@ public class CompositeFuture<T> implements Future<T> {
		List<T> results = new ArrayList<>(this.futures.size());
		for (Future<T> future : this.futures) {
		if (future instanceof RunnableFuture<?>) {
		this.logger.debug("Waiting {} ms since the start of the execution of the future to complete", unit.toMillis(timeout));
		this.logger.trace("Waiting {} ms since the start of the exectuion of the future to complete", unit.toMillis(timeout));
		results.add(((RunnableFuture<T>) future).get(timeout, unit));
		this.logger.trace("Exectuion completed", unit.toMillis(timeout));
		} else {
		long remainingTimeMillis = expireTimeMillis - System.currentTimeMillis();
		this.logger.debug("Waiting {} ms for the future to complete", remainingTimeMillis);
		this.logger.trace("Waiting {} ms for the future to complete", remainingTimeMillis);
		results.add(future.get(remainingTimeMillis, TimeUnit.MILLISECONDS));
		this.logger.trace("Exectuion completed", unit.toMillis(timeout));
		}
		}

		@@ -131,14 +133,24 @@ public class CompositeFuture<T> implements Future<T> {
		Future<T> future = i.next();
		if (future.isCancelled()) {
		if (includeCancelled) {
		this.logger.trace("Removing cancelled future");
		removedCancelled++;
		i.remove();
		} else {
		remain++;
		}
		} else if (future.isDone()) {
		try {
		future.get();
		} catch (InterruptedException ie) {
		this.logger.trace("Future completed because it was interrupted");
		} catch (ExecutionException ee) {
		this.logger.error(ee.getMessage(), ee);
		} finally {
		this.logger.trace("Removing completed future");
		removedDone++;
		i.remove();
		}
		} else if (future instanceof CompositeFuture<?>) {
		cfutures.add((CompositeFuture<?>) future);
		} else {

shared/src/main/resources/alfresco/module/com_inteligr8_alfresco_asie-shared/alfresco-global.properties

+2 −0

Original line number	Diff line number	Diff line
		@@ -19,6 +19,8 @@ inteligr8.asie.reconciliation.nodesChunkSize=250
		inteligr8.asie.reconciliation.nodeTimeoutSeconds=10
		inteligr8.asie.reconciliation.concurrentQueueSize=32
		inteligr8.asie.reconciliation.concurrency=2
		inteligr8.asie.reconciliation.waitAfterSolrNodeActionMillis=0
		inteligr8.asie.reconciliation.waitAfterSolrNodeReconcileMillis=0

		# Action (like indexing and re-indexing) configuration
		inteligr8.asie.default.concurrentQueueSize=32

shared/src/main/java/com/inteligr8/alfresco/asie/service/ExecutorManager.java

+1 −1

File changed.

Contains only whitespace changes.

shared/src/main/java/com/inteligr8/alfresco/asie/service/SolrShardHashService.java

+1 −1

File changed.

Contains only whitespace changes.