From 66933e587d84f03c312463c7fababdb13694b98c Mon Sep 17 00:00:00 2001 From: "Brian M. Long" Date: Mon, 2 Feb 2026 18:48:33 -0500 Subject: [PATCH] separate reconcile threads from index/reindex threads --- .../service/AbstractNodeActionService.java | 24 +++- .../asie/service/AcsReconcileService.java | 130 +++++++++--------- .../asie/service/ExecutorManager.java | 58 +++----- .../util/ThrottledThreadPoolExecutor.java | 3 - 4 files changed, 100 insertions(+), 115 deletions(-) diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java b/shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java index db1dd10..db8012f 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java +++ b/shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java @@ -43,7 +43,7 @@ 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()); @@ -81,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 @@ -140,11 +156,7 @@ public abstract class AbstractNodeActionService { this.logger.debug("Will attempt to {} ACS node against {} shard instances: {}", this.getActionName(), eligibleInstances.size(), nodeDbId); CompositeFuture 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); diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java b/shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java index d3dbdf2..878a101 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java +++ b/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,6 +60,9 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @Autowired private ReindexService reindexService; + @Autowired + private ExecutorManager executorManager; + @Value("${inteligr8.asie.reconciliation.nodesChunkSize}") private int nodesChunkSize; @@ -74,17 +75,20 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @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 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() { @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 syncHosts = new HashSet<>(); - Set asyncHosts = new HashSet<>(); - Map errorHosts = new HashMap<>(); - + public Future 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 syncHosts = new HashSet<>(); - Set asyncHosts = new HashSet<>(); - Map errorHosts = new HashMap<>(); - + public Future 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) { diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/service/ExecutorManager.java b/shared/src/main/java/com/inteligr8/alfresco/asie/service/ExecutorManager.java index aa821c4..2028cc5 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/service/ExecutorManager.java +++ b/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 { + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + @Value("${inteligr8.asie.executors.expireTimeInMinutes}") private int expireTimeInMinutes; - private Cache refCache; - private Cache expiringCache; + private Cache 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 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() { + return this.cache.get(name, new Callable() { @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() { - @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); } } diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/util/ThrottledThreadPoolExecutor.java b/shared/src/main/java/com/inteligr8/alfresco/asie/util/ThrottledThreadPoolExecutor.java index d36d383..be921bf 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/util/ThrottledThreadPoolExecutor.java +++ b/shared/src/main/java/com/inteligr8/alfresco/asie/util/ThrottledThreadPoolExecutor.java @@ -80,9 +80,6 @@ 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 {