v1.3.5 poms

asie-api/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>com.inteligr8.alfresco</groupId>
 		<artifactId>asie-platform-module-parent</artifactId>
-		<version>1.3.3</version>
+		<version>1.3.5</version>
 		<relativePath>../</relativePath>
 	</parent>
 

enterprise-module/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>com.inteligr8.alfresco</groupId>
 		<artifactId>asie-platform-module-parent</artifactId>
-		<version>1.3.3</version>
+		<version>1.3.5</version>
 		<relativePath>../</relativePath>
 	</parent>
 

pom.xml

@@ -5,7 +5,7 @@
 	
 	<groupId>com.inteligr8.alfresco</groupId>
 	<artifactId>asie-platform-module-parent</artifactId>
-	<version>1.3.3</version>
+	<version>1.3.5</version>
 	<packaging>pom</packaging>
 	
 	<name>ASIE Platform Module Parent</name>

shared/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>com.inteligr8.alfresco</groupId>
 		<artifactId>asie-platform-module-parent</artifactId>
-		<version>1.3.3</version>
+		<version>1.3.5</version>
 		<relativePath>../</relativePath>
 	</parent>
 

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

@@ -9,14 +9,8 @@ import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import org.alfresco.model.ContentModel;
-import org.alfresco.service.cmr.repository.InvalidNodeRefException;
-import org.alfresco.service.cmr.repository.NodeRef;
-import org.alfresco.service.cmr.repository.NodeService;
-import org.alfresco.service.cmr.repository.StoreRef;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.extensions.webscripts.WebScriptException;
 import org.springframework.extensions.webscripts.WebScriptRequest;
 import org.springframework.extensions.webscripts.WebScriptResponse;

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

@@ -12,17 +12,12 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import org.alfresco.model.ContentModel;
 import org.alfresco.repo.index.shard.Floc;
 import org.alfresco.repo.index.shard.Shard;
 import org.alfresco.repo.index.shard.ShardInstance;
 import org.alfresco.repo.index.shard.ShardRegistry;
 import org.alfresco.repo.index.shard.ShardState;
 import org.alfresco.service.cmr.repository.StoreRef;
-import org.alfresco.service.cmr.search.SearchParameters;
-import org.alfresco.service.cmr.search.SearchService;
-import org.alfresco.service.namespace.NamespaceService;
-import org.alfresco.service.namespace.QName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -45,9 +40,6 @@ public abstract class AbstractActionService {
 
 	private final Logger logger = LoggerFactory.getLogger(this.getClass());
 	
-	@Autowired
-	private NamespaceService namespaceService;
-	
 	@Autowired
 	private ApiService apiService;
 	
@@ -58,10 +50,10 @@ public abstract class AbstractActionService {
 	@Qualifier(Constants.QUALIFIER_ASIE)
 	private ShardRegistry shardRegistry;
 	
-	@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
+	@Value("${inteligr8.asie.default.concurrentQueueSize}")
 	private int concurrentQueueSize;
 	
-	@Value("${inteligr8.asie.default.concurrency:16}")
+	@Value("${inteligr8.asie.default.concurrency}")
 	private int concurrency;
 	
 	protected int getConcurrency() {

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

@@ -8,10 +8,10 @@ import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.regex.Matcher;
 
 import org.alfresco.model.ContentModel;
 import org.alfresco.repo.index.shard.Floc;
@@ -19,8 +19,6 @@ import org.alfresco.repo.index.shard.Shard;
 import org.alfresco.repo.index.shard.ShardInstance;
 import org.alfresco.repo.index.shard.ShardRegistry;
 import org.alfresco.repo.index.shard.ShardState;
-import org.alfresco.service.cmr.repository.NodeRef;
-import org.alfresco.service.cmr.repository.NodeService;
 import org.alfresco.service.cmr.repository.StoreRef;
 import org.alfresco.service.cmr.search.SearchParameters;
 import org.alfresco.service.cmr.search.SearchService;
@@ -28,6 +26,7 @@ import org.alfresco.service.namespace.NamespaceService;
 import org.alfresco.service.namespace.QName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
@@ -44,16 +43,13 @@ import com.inteligr8.solr.model.Action;
 import com.inteligr8.solr.model.ActionResponse;
 import com.inteligr8.solr.model.BaseResponse;
 
-public abstract class AbstractNodeActionService {
+public abstract class AbstractNodeActionService implements DisposableBean {
 
 	private final Logger logger = LoggerFactory.getLogger(this.getClass());
 	
 	@Autowired
 	private NamespaceService namespaceService;
 	
-	@Autowired
-	private NodeService nodeService;
-	
 	@Autowired
 	private ApiService apiService;
 	
@@ -67,10 +63,10 @@ public abstract class AbstractNodeActionService {
 	@Qualifier(Constants.QUALIFIER_ASIE)
 	private ShardRegistry shardRegistry;
 	
-	@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
+	@Value("${inteligr8.asie.default.concurrentQueueSize}")
 	private int concurrentQueueSize;
 	
-	@Value("${inteligr8.asie.default.concurrency:16}")
+	@Value("${inteligr8.asie.default.concurrency}")
 	private int concurrency;
 	
 	protected int getConcurrency() {
@@ -85,6 +81,22 @@ public abstract class AbstractNodeActionService {
 	
 	protected abstract String getActionName();
 	
+	@Override
+	public void destroy() {
+		ExecutorService executor = this.executorManager.get(this.getThreadNamePrefix());
+		if (executor != null) {
+			this.logger.info("Shutting down throttled thread pool executor: {}", this.getThreadNamePrefix());
+			executor.shutdown();
+		}
+	}
+
+	private ThrottledThreadPoolExecutor getExecutor() {
+		return this.executorManager.createThrottled(
+				this.getThreadNamePrefix(),
+				this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),
+				1L, TimeUnit.MINUTES);
+	}
+	
 	/**
 	 * This method executes an action on the specified node in Solr using its
 	 * ACS unique database identifier.  The callback handles all the return
@@ -144,11 +156,7 @@ public abstract class AbstractNodeActionService {
 		this.logger.debug("Will attempt to {} ACS node against {} shard instances: {}", this.getActionName(), eligibleInstances.size(), nodeDbId);
 		
 		CompositeFuture<Void> future = new CompositeFuture<>();
-		
-		ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(
-				this.getThreadNamePrefix(),
-				this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),
-				1L, TimeUnit.MINUTES);
+		ThrottledThreadPoolExecutor executor = this.getExecutor();
 		
 		for (final com.inteligr8.alfresco.asie.model.ShardInstance instance : eligibleInstances) {
 			this.logger.trace("Will attempt to {} ACS node against shard instance: {}: {}", this.getActionName(), nodeDbId, instance);
@@ -233,7 +241,7 @@ public abstract class AbstractNodeActionService {
 			if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE))
 				continue;
 
-			Integer shardHash = null;
+			int shardHash = -1;
 			ShardSet shardset = null;
 			for (Entry<Shard, Set<ShardState>> shard : floc.getValue().entrySet()) {
 				for (ShardState shardState : shard.getValue()) {
@@ -242,26 +250,10 @@ public abstract class AbstractNodeActionService {
 					// so we are computing the shardHash once and caching it
 					if (shardset == null) {
 						shardset = ShardSet.from(floc.getKey(), shardState);
-						switch (shardset.getMethod()) {
-							case PROPERTY:
-								this.logger.trace("Using property-based sharding method; discovering target shard ...");
-								NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
-								Object propValue = this.nodeService.getProperty(nodeRef, QName.createQName(shardset.getPrefixedProperty(), this.namespaceService));
-								if (propValue != null) {
-									this.logger.trace("Discovered node property for sharding: {} <=> {} => {}", nodeDbId, nodeRef, propValue);
-									Matcher matcher = shardset.getRegex().matcher(propValue.toString());
-									String hashable = matcher.group(1);
-									this.logger.trace("Extracted shardable value from node: {} <=> {} => {}", nodeDbId, propValue, hashable);
-									shardHash = this.shardHashService.hash(hashable, shardset.getShards().intValue());
-									this.logger.debug("Hash shardable value to shard instance ID: {} <=> {} => {}", nodeDbId, hashable, shardHash);
-								}
-								break;
-							default:
-								this.logger.trace("Despite sharding, considering all shards and nodes without optimization: {}: {}", nodeDbId, instances);
-						}
+						shardHash = this.shardHashService.computeShardInstanceId(shardset, nodeDbId);
 					}
 					
-					if (shardHash != null) {
+					if (shardHash >= 0) {
 						if (shard.getKey().getInstance() == shardHash)
 							instances.add(this.toModel(shardState.getShardInstance(), shardState));
 					} else {

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

@@ -1,11 +1,10 @@
 package com.inteligr8.alfresco.asie.service;
 
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
+import java.util.Collections;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -26,7 +25,6 @@ import org.apache.commons.collections4.SetUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
@@ -39,7 +37,7 @@ import com.inteligr8.alfresco.asie.util.CompositeFuture;
 import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
 
 @Component
-public class AcsReconcileService implements InitializingBean, DisposableBean {
+public class AcsReconcileService implements DisposableBean {
 
 	private final Logger logger = LoggerFactory.getLogger(this.getClass());
 	private final Logger reconcileLogger = LoggerFactory.getLogger("inteligr8.asie.reconcile");
@@ -62,29 +60,35 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 	@Autowired
 	private ReindexService reindexService;
 	
-	@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
+	@Autowired
+	private ExecutorManager executorManager;
+	
+	@Value("${inteligr8.asie.reconciliation.nodesChunkSize}")
 	private int nodesChunkSize;
 	
-	@Value("${inteligr8.asie.reconciliation.nodeTimeoutSeconds:10}")
+	@Value("${inteligr8.asie.reconciliation.nodeTimeoutSeconds}")
 	private int nodeTimeoutSeconds;
 	
-	@Value("${inteligr8.asie.reconciliation.concurrentQueueSize:64}")
+	@Value("${inteligr8.asie.reconciliation.concurrentQueueSize}")
 	private int concurrentQueueSize;
 	
-	@Value("${inteligr8.asie.reconciliation.concurrency:2}")
+	@Value("${inteligr8.asie.reconciliation.concurrency}")
 	private int concurrency;
 	
-	private ThrottledThreadPoolExecutor executor;
-	
-	@Override
-	public void afterPropertiesSet() {
-		this.executor = new ThrottledThreadPoolExecutor(this.concurrency, this.concurrency, this.concurrentQueueSize, 1L, TimeUnit.MINUTES, "solr-reconcile");
-		this.executor.prestartAllCoreThreads();
-	}
-	
 	@Override
 	public void destroy() {
-		this.executor.shutdown();
+		ExecutorService executor = this.executorManager.get("solr-reconcile");
+		if (executor != null) {
+			this.logger.info("Shutting down throttled thread pool executor: {}", "solr-reconcile");
+			executor.shutdown();
+		}
+	}
+
+	private ThrottledThreadPoolExecutor getExecutor() {
+		return this.executorManager.createThrottled(
+				"solr-reconcile",
+				this.concurrency, this.concurrency, this.concurrentQueueSize,
+				10L, TimeUnit.SECONDS);
 	}
 	
 	/**
@@ -205,6 +209,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 		}
 		
 		CompositeFuture<Void> future = new CompositeFuture<>();
+		ThrottledThreadPoolExecutor executor = this.getExecutor();
 
 		for (long _nodeDbId = fromDbId; _nodeDbId < toDbId; _nodeDbId++) {
 			final long nodeDbId = _nodeDbId;
@@ -222,7 +227,11 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 						callback.reconciled(nodeDbId);
 
 						if (reindexReconciled)
-							reindex(nodeDbId, nodeRefs[dbIdIndex], callback, execTimeout, execUnit);
+							reindex(nodeDbId, nodeRefs[dbIdIndex], callback);
+							// purposefully forgetting about the returned future
+							// its results will be logged
+							// the reconcile thread will continue independently
+							// the callback will lag
 						return null;
 					}
 				};
@@ -230,16 +239,16 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 				callable = new Callable<Void>() {
 					@Override
 					public Void call() throws InterruptedException, TimeoutException {
-						reconcile(nodeDbId, indexUnreconciled, callback, execTimeout, execUnit);
+						reconcile(nodeDbId, indexUnreconciled, callback);
 						return null;
 					}
 				};
 			}
 			
 			if (queueTimeout < 0L) {
-				future.combine(this.executor.submit(callable, -1L, null));
+				future.combine(executor.submit(callable, -1L, null));
 			} else {
-				future.combine(this.executor.submit(callable, queueTimeout, queueUnit));
+				future.combine(executor.submit(callable, queueTimeout, queueUnit));
 			}
 		}
 		
@@ -248,8 +257,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 	
 	public void reconcile(long nodeDbId,
 			boolean index,
-			ReconcileCallback callback,
-			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
+			ReconcileCallback callback) throws InterruptedException, TimeoutException {
 		NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
 		if (nodeRef == null) {
 			this.logger.trace("No such ACS node: {}; skipping ...", nodeDbId);
@@ -273,93 +281,81 @@ public class AcsReconcileService implements InitializingBean, DisposableBean {
 			this.reconcileLogger.info("UNRECONCILED: {} <=> {}", nodeDbId, nodeRef);
 			callback.unreconciled(nodeDbId);
 		} else {
-			logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);
-			this.index(nodeDbId, nodeRef, callback, execTimeout, execUnit);
+			this.logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);
+			this.index(nodeDbId, nodeRef, callback);
+			// purposefully forgetting about the returned future
+			// its results will be logged
+			// the reconcile thread will continue independently
+			// the callback will lag
 		}
 	}
 	
-	public void index(long nodeDbId, NodeRef nodeRef,
-			ReconcileCallback callback,
-			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
-		Set<ShardInstance> syncHosts = new HashSet<>();
-		Set<ShardInstance> asyncHosts = new HashSet<>();
-		Map<ShardInstance, String> errorHosts = new HashMap<>();
-		
+	public Future<Void> index(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException, TimeoutException {
 		IndexCallback indexCallback = new IndexCallback() {
 			
 			@Override
 			public void success(ShardInstance instance) {
 				reconcileLogger.info("INDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				syncHosts.add(instance);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.singleton(instance), Collections.emptySet(),
+						Collections.emptyMap());
 			}
 			
 			@Override
 			public void scheduled(ShardInstance instance) {
 				reconcileLogger.info("INDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				asyncHosts.add(instance);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.emptySet(), Collections.singleton(instance),
+						Collections.emptyMap());
 			}
 			
 			@Override
 			public void error(ShardInstance instance, String message) {
 				reconcileLogger.info("FAILED INDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				errorHosts.put(instance, message);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.emptySet(), Collections.emptySet(),
+						Collections.singletonMap(instance, message));
 			}
 		};
 		
-		try {
-			if (execTimeout < 0L) {
-				this.indexService.index(nodeDbId, indexCallback).get();
-			} else {
-				this.indexService.index(nodeDbId, indexCallback).get(execTimeout, execUnit);
-			}
-		} catch (ExecutionException ee) {
-			throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);
-		}
-		
-		if (callback != null)
-			callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);
+		return this.indexService.index(nodeDbId, indexCallback);
 	}
 	
-	public void reindex(long nodeDbId, NodeRef nodeRef,
-			ReconcileCallback callback,
-			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
-		Set<ShardInstance> syncHosts = new HashSet<>();
-		Set<ShardInstance> asyncHosts = new HashSet<>();
-		Map<ShardInstance, String> errorHosts = new HashMap<>();
-		
+	public Future<Void> reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException {
 		ReindexCallback reindexCallback = new ReindexCallback() {
 			
 			@Override
 			public void success(ShardInstance instance) {
 				reconcileLogger.info("REINDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				syncHosts.add(instance);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.singleton(instance), Collections.emptySet(),
+						Collections.emptyMap());
 			}
 			
 			@Override
 			public void scheduled(ShardInstance instance) {
 				reconcileLogger.info("REINDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				asyncHosts.add(instance);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.emptySet(), Collections.singleton(instance),
+						Collections.emptyMap());
 			}
 			
 			@Override
 			public void error(ShardInstance instance, String message) {
 				reconcileLogger.info("FAILED REINDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance);
-				errorHosts.put(instance, message);
+				if (callback != null)
+					callback.processed(nodeDbId,
+						Collections.emptySet(), Collections.emptySet(),
+						Collections.singletonMap(instance, message));
 			}
 		};
-		
-		try {
-			if (execTimeout < 0L) {
-				this.reindexService.reindex(nodeDbId, reindexCallback).get();
-			} else {
-				this.reindexService.reindex(nodeDbId, reindexCallback).get(execTimeout, execUnit);
-			}
-		} catch (ExecutionException ee) {
-			throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);
-		}
-		
-		if (callback != null)
-			callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);
+
+		return this.reindexService.reindex(nodeDbId, reindexCallback);
 	}
 	
 	private String formatForFts(QName qname) {

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

@@ -43,7 +43,7 @@ public class ApiService implements InitializingBean {
 	@Value("${inteligr8.asie.basePath}")
 	private String solrBaseUrl;
 	
-	@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
+	@Value("${inteligr8.asie.reconciliation.nodesChunkSize}")
 	private int nodesChunkSize;
 	
 	@Override

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

@@ -6,6 +6,8 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.TimeUnit;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Value;
@@ -31,41 +33,35 @@ import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
 @Component
 public class ExecutorManager implements InitializingBean, DisposableBean, RemovalListener<String, ExecutorService> {
 	
-	@Value("${inteligr8.asie.executors.expireTimeInMinutes:30}")
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	@Value("${inteligr8.asie.executors.expireTimeInMinutes}")
 	private int expireTimeInMinutes;
 	
-	private Cache<String, ExecutorService> refCache;
-	private Cache<String, ExecutorService> expiringCache;
+	private Cache<String, ThrottledThreadPoolExecutor> cache;
 	
 	@Override
 	public void afterPropertiesSet() throws Exception {
-		// a weak value happens when the executor is no longer referenced
-		// the possible references are by the caller temporarily using and the `expiringCache` (below; so it expired)
-		// this keeps the pool from being shutdown after it expires if the caller is still referencing it
-		// ultimately, if it is cached, it will be in this cache and MAY be in the `expiringCache`. 
-		this.refCache = CacheBuilder.newBuilder()
-				.initialCapacity(8)
-				.weakValues()
-				.removalListener(this)
-				.build();
-		
-		this.expiringCache = CacheBuilder.newBuilder()
+		this.cache = CacheBuilder.newBuilder()
 				.initialCapacity(8)
 				.expireAfterAccess(this.expireTimeInMinutes, TimeUnit.MINUTES)
+				.removalListener(this)
 				.build();
 	}
 	
 	@Override
 	public void destroy() throws Exception {
-		this.refCache.invalidateAll();
-		this.refCache.cleanUp();
-		this.expiringCache.invalidateAll();
-		this.expiringCache.cleanUp();
+		this.cache.invalidateAll();
+		this.cache.cleanUp();
 	}
 	
 	@Override
 	public void onRemoval(RemovalNotification<String, ExecutorService> notification) {
-		notification.getValue().shutdown();
+		this.logger.debug("Throttled thread pool removed/expired from cache: {}", notification.getKey());
+		if (!notification.getValue().isShutdown()) {
+			notification.getValue().shutdown();
+			this.logger.info("Throttled thread pool shut down: {}", notification.getKey());
+		}
 	}
 	
 	public ThrottledThreadPoolExecutor createThrottled(
@@ -88,9 +84,10 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
             final RejectedExecutionHandler rejectedExecutionHandler) {
 		try {
 			// if it is already cached, reuse the cache; otherwise create one
-			final ExecutorService executor = this.refCache.get(name, new Callable<ThrottledThreadPoolExecutor>() {
+			return this.cache.get(name, new Callable<ThrottledThreadPoolExecutor>() {
 				@Override
 				public ThrottledThreadPoolExecutor call() {
+					logger.info("Creating throttled thread pool: {}", name);
 					ThrottledThreadPoolExecutor executor = null;
 					if (rejectedExecutionHandler == null) {
 						executor = new ThrottledThreadPoolExecutor(coreThreadPoolSize, maximumThreadPoolSize, maximumQueueSize,
@@ -103,14 +100,9 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
 								rejectedExecutionHandler);
 					}
 					
+					logger.debug("Created throttled thread pool: {}; threads: {}; queue: {}", name, maximumThreadPoolSize, maximumQueueSize);
 					executor.prestartAllCoreThreads();
-					return executor;
-				}
-			});
-			
-			return (ThrottledThreadPoolExecutor) this.expiringCache.get(name, new Callable<ExecutorService>() {
-				@Override
-				public ExecutorService call() throws Exception {
+					logger.trace("Started {} core threads in thread pool: {}", coreThreadPoolSize, name);
 					return executor;
 				}
 			});
@@ -120,19 +112,7 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
 	}
 	
 	public ExecutorService get(String name) {
-		// grab from the expiring cache first, so we can 
-		ExecutorService executor = this.expiringCache.getIfPresent(name);
-		if (executor != null)
-			return executor;
-		
-		executor = this.refCache.getIfPresent(name);
-		if (executor == null)
-			return null;
-		
-		// the executor expired, but it was still referenced by the caller
-		// re-cache it
-		this.expiringCache.put(name, executor);
-		return executor;
+		return this.cache.getIfPresent(name);
 	}
 	
 }

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

@@ -1,14 +1,32 @@
 package com.inteligr8.alfresco.asie.service;
 
 import java.nio.charset.Charset;
+import java.util.regex.Matcher;
 
+import org.alfresco.error.AlfrescoRuntimeException;
+import org.alfresco.service.cmr.repository.NodeRef;
+import org.alfresco.service.cmr.repository.NodeService;
+import org.alfresco.service.namespace.NamespaceService;
+import org.alfresco.service.namespace.QName;
 import org.apache.commons.codec.digest.MurmurHash3;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import com.inteligr8.alfresco.asie.model.ShardSet;
+
 @Component
 public class SolrShardHashService {
 	
 	private final Charset charset = Charset.forName("utf-8");
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+	
+	@Autowired
+	private NamespaceService namespaceService;
+	
+	@Autowired
+	private NodeService nodeService;
 	
 	public int hash(Object obj, int shardCount) {
 		String str = obj.toString();
@@ -23,5 +41,53 @@ public class SolrShardHashService {
 		hash.add(bytes, 0, bytes.length);
 		return Math.abs(hash.end()) % shardCount;
 	}
+	
+	/**
+	 * @param shardset A ShardSet.
+	 * @param nodeDbId The numeric database ID of an ACS node.
+	 * @return A shard instance from 0 to 1 less than the number of shards; -1 if unable to hash.
+	 */
+	public int computeShardInstanceId(ShardSet shardset, long nodeDbId) {
+		NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
+		if (nodeRef == null)
+			throw new AlfrescoRuntimeException("The node " + nodeDbId + " does not exist");
+		return this.computeShardInstanceId(shardset, nodeRef);
+	}
+
+	/**
+	 * @param shardset A ShardSet.
+	 * @param nodeRef A reference to an ACS node.
+	 * @return A shard instance from 0 to 1 less than the number of shards; -1 if unable to hash.
+	 */
+	public int computeShardInstanceId(ShardSet shardset, NodeRef nodeRef) {
+		switch (shardset.getMethod()) {
+			case PROPERTY:
+				QName hashableProperty = QName.createQName(shardset.getPrefixedProperty(), this.namespaceService);
+				Object fullPropertyValue = this.nodeService.getProperty(nodeRef, hashableProperty);
+				if (fullPropertyValue == null) {
+					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 = fullPropertyValue.toString();
+				if (shardset.getRegex() != null) {
+					Matcher matcher = shardset.getRegex().matcher(hashableValue);
+					if (!matcher.find()) {
+						this.logger.debug("Unable to determine shard instance ID because regex pattern doesn't match the hashing property value: {}: {}: {}", nodeRef, shardset.getRegex(), hashableValue);
+						return -1;
+					}
+					hashableValue = matcher.group(1);
+					this.logger.trace("Extracted shardable value from node: {}: {} => {}", nodeRef, fullPropertyValue, hashableValue);
+				}
+				
+				int shardHash = this.hash(hashableValue, shardset.getShards().intValue());
+				this.logger.debug("Hash shardable value to shard instance ID: {}: {} => {}", nodeRef, hashableValue, shardHash);
+				return shardHash;
+			default:
+				this.logger.trace("Unable to determine shard instance ID due to shard method: {}", shardset.getMethod());
+				return -1;
+		}
+	}
 
 }

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

@@ -79,7 +79,7 @@ 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 exectuion of the future to complete", unit.toMillis(timeout));
+				this.logger.debug("Waiting {} ms since the start of the execution of the future to complete", unit.toMillis(timeout));
 				results.add(((RunnableFuture<T>) future).get(timeout, unit));
 			} else {
 				long remainingTimeMillis = expireTimeMillis - System.currentTimeMillis();
@@ -124,6 +124,7 @@ public class CompositeFuture<T> implements Future<T> {
 		List<CompositeFuture<?>> cfutures = new LinkedList<>();
 		int removedCancelled = 0;
 		int removedDone = 0;
+		int remain = 0;
 		
 		Iterator<Future<T>> i = this.futures.iterator();
 		while (i.hasNext()) {
@@ -132,16 +133,20 @@ public class CompositeFuture<T> implements Future<T> {
 				if (includeCancelled) {
 					removedCancelled++;
 					i.remove();
+				} else {
+					remain++;
 				}
 			} else if (future.isDone()) {
 				removedDone++;
 				i.remove();
 			} else if (future instanceof CompositeFuture<?>) {
 				cfutures.add((CompositeFuture<?>) future);
+			} else {
+				remain++;
 			}
 		}
 		
-		this.logger.debug("Purged {} cancelled and {} completed futures", removedCancelled, removedDone);
+		this.logger.debug("Purged {} cancelled and {} completed futures; {} remain", removedCancelled, removedDone, remain);
 		
 		for (CompositeFuture<?> cfuture : cfutures)
 			cfuture.purge(includeCancelled);

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

@@ -80,14 +80,11 @@ public class ThrottledThreadPoolExecutor extends ThreadPoolExecutor {
 	}
 	
 	private WaitableRunnable submit(WaitableRunnable runnable, long throttlingBlockTimeout, TimeUnit throttlingBlockUnit) throws InterruptedException, TimeoutException {
-		// if no core threads are running, the queue won't be monitored for runnables
-		this.prestartAllCoreThreads();
-
 		if (throttlingBlockTimeout < 0L) {
 			this.getQueue().put(runnable);
 		} else {
 			if (!this.getQueue().offer(runnable, throttlingBlockTimeout, throttlingBlockUnit))
-				throw new TimeoutException();
+				throw new TimeoutException("Timeout waiting for queue space for runnable");
 		}
 		
 		return runnable;

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

@@ -1,8 +1,25 @@
 
 # defaulting to 3 days = 60 * 24 * 3 = 4320
+# once the node is selected, no other node for the shard will be used for the backup
 inteligr8.asie.backup.persistTimeMinutes=4320
 
+# what authorities (users or groups) may use the REST services provided by this module?
 inteligr8.asie.allowedAuthorities=GROUP_ALFRESCO_ADMINISTRATORS
 
 # same as solr.baseUrl, but that property is private to the Search subsystem
 inteligr8.asie.basePath=/solr
+
+
+# How long should idle executors remain before being shutdown?
+# They will re-initialize if needed again
+inteligr8.asie.executors.expireTimeInMinutes=30
+
+# Reconciliation configuration; each node will be processed in its own thread
+inteligr8.asie.reconciliation.nodesChunkSize=250
+inteligr8.asie.reconciliation.nodeTimeoutSeconds=10
+inteligr8.asie.reconciliation.concurrentQueueSize=32
+inteligr8.asie.reconciliation.concurrency=2
+
+# Action (like indexing and re-indexing) configuration
+inteligr8.asie.default.concurrentQueueSize=32
+inteligr8.asie.default.concurrency=2

solr-api/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>com.inteligr8.alfresco</groupId>
 		<artifactId>asie-platform-module-parent</artifactId>
-		<version>1.3.3</version>
+		<version>1.3.5</version>
 		<relativePath>../</relativePath>
 	</parent>