various minor improvements

(cherry picked from commit f2cf774bad5987c91b4e8acef94fb97fdb921063)

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

@@ -232,10 +232,6 @@ public abstract class AbstractNodeActionService implements DisposableBean {
 	private List<com.inteligr8.alfresco.asie.model.ShardInstance> findPossibleShardInstances(long nodeDbId) {
 		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<>();
 		

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

@@ -59,7 +59,7 @@ public class AcsReconcileService implements DisposableBean {
 	
 	@Autowired
 	private ReindexService reindexService;
-	
+
 	@Autowired
 	private ExecutorManager executorManager;
 	
@@ -100,7 +100,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 +111,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 +123,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) {
+		Future<Void> future = this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit);
-			long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
-			future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));
-			future.purge(true);
-		}
 		
 		try {
 			future.get(execTimeout, execUnit);
@@ -143,27 +135,52 @@ 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 {
-		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) {
+			return this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null);
-				long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
-				future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null));
-				future.purge(true);
-			}
 		} 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;
 		
+		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);
+		}
+
 		return future;
 	}
 	
@@ -211,6 +228,11 @@ 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);

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

@@ -32,7 +32,7 @@ import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
  */
 @Component
 public class ExecutorManager implements InitializingBean, DisposableBean, RemovalListener<String, ExecutorService> {
-	
+
 	private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
 	@Value("${inteligr8.asie.executors.expireTimeInMinutes}")

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

@@ -72,7 +72,7 @@ public class SolrShardHashService {
 					this.logger.debug("Unable to determine shard instance ID because property does not exist on node: {}: {}", nodeRef, hashableProperty);
 					return -1;
 				}
-				
+
 				this.logger.trace("Discovered node property for sharding: {} => {}", nodeRef, fullPropertyValue);
 
 				String hashableValue = null;

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

@@ -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()) {
-				removedDone++;
+				try {
-				i.remove();
+					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 {