From 06d16eb223e7dcb5cf7eb22b98235b9d8c0b9165 Mon Sep 17 00:00:00 2001 From: Brian Long Date: Mon, 12 Jan 2026 19:31:26 -0500 Subject: [PATCH 1/3] various minor improvements (cherry picked from commit f2cf774bad5987c91b4e8acef94fb97fdb921063) --- .../service/AbstractNodeActionService.java | 4 -- .../asie/service/AcsReconcileService.java | 66 ++++++++++++------- .../asie/service/ExecutorManager.java | 2 +- .../asie/service/SolrShardHashService.java | 2 +- .../alfresco/asie/util/CompositeFuture.java | 20 ++++-- 5 files changed, 62 insertions(+), 32 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 2ce701e..7c324a3 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 @@ -232,10 +232,6 @@ public abstract class AbstractNodeActionService implements DisposableBean { private List 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 instances = new LinkedList<>(); 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 878a101..f5ca15f 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 @@ -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 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 future = this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit); 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 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 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.purge(true); - } + 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 _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 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 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); 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 2028cc5..207f9eb 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 @@ -32,7 +32,7 @@ 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}") diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/service/SolrShardHashService.java b/shared/src/main/java/com/inteligr8/alfresco/asie/service/SolrShardHashService.java index 972ba76..f62af22 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/service/SolrShardHashService.java +++ b/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; diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/util/CompositeFuture.java b/shared/src/main/java/com/inteligr8/alfresco/asie/util/CompositeFuture.java index d575317..174f439 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/util/CompositeFuture.java +++ b/shared/src/main/java/com/inteligr8/alfresco/asie/util/CompositeFuture.java @@ -79,12 +79,14 @@ public class CompositeFuture implements Future { List results = new ArrayList<>(this.futures.size()); for (Future 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) 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 implements Future { Future future = i.next(); if (future.isCancelled()) { if (includeCancelled) { + this.logger.trace("Removing cancelled future"); removedCancelled++; i.remove(); } else { remain++; } } else if (future.isDone()) { - removedDone++; - i.remove(); + 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 { From c0b02b900472921be9bb8b456b2a82cdf68730db Mon Sep 17 00:00:00 2001 From: Brian Long Date: Mon, 12 Jan 2026 20:17:00 -0500 Subject: [PATCH 2/3] basic cleanup (cherry picked from commit 727a566ad55100023d613b98bb69f7ee21891be3) --- .../alfresco/asie/rest/ReconcileAcsNodesWebScript.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java b/shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java index f2b7c96..66917df 100644 --- a/shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java +++ b/shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java @@ -44,10 +44,10 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript { public void reconciled(long nodeDbId) { if (includeReconciled) { @SuppressWarnings("unchecked") - List unreconciledNodeDbIds = (List) responseMap.get("reconciled"); - if (unreconciledNodeDbIds == null) - responseMap.put("reconciled", unreconciledNodeDbIds = new LinkedList<>()); - unreconciledNodeDbIds.add(nodeDbId); + List reconciledNodeDbIds = (List) responseMap.get("reconciled"); + if (reconciledNodeDbIds == null) + responseMap.put("reconciled", reconciledNodeDbIds = new LinkedList<>()); + reconciledNodeDbIds.add(nodeDbId); } } From 0685bf9e762d60819537c0fbbf6e0234ec50f055 Mon Sep 17 00:00:00 2001 From: Brian Long Date: Tue, 24 Mar 2026 22:30:58 -0400 Subject: [PATCH 3/3] add reconcile throttling waits --- .../asie/service/AcsReconcileService.java | 41 ++++++++++++++++--- .../alfresco-global.properties | 2 + 2 files changed, 37 insertions(+), 6 deletions(-) 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 f5ca15f..8fb4ee0 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 @@ -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"); @@ -238,6 +244,7 @@ public class AcsReconcileService implements DisposableBean { this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId); Callable callable; + boolean callingSolr = false; final int dbIdIndex = (int) (nodeDbId - fromDbId); if (nodeRefs[dbIdIndex] != null) { @@ -257,6 +264,8 @@ public class AcsReconcileService implements DisposableBean { return null; } }; + if (reindexReconciled) + callingSolr = true; } else { callable = new Callable() { @Override @@ -265,6 +274,7 @@ public class AcsReconcileService implements DisposableBean { return null; } }; + callingSolr = true; } if (queueTimeout < 0L) { @@ -272,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 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); @@ -309,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; } } @@ -343,7 +360,13 @@ public class AcsReconcileService implements DisposableBean { } }; - return this.indexService.index(nodeDbId, indexCallback); + Future future = this.indexService.index(nodeDbId, indexCallback); + + if (this.waitAfterSolrNodeActionMillis > 0L) { + Thread.sleep(this.waitAfterSolrNodeActionMillis); + } + + return future; } public Future reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException { @@ -377,7 +400,13 @@ public class AcsReconcileService implements DisposableBean { } }; - return this.reindexService.reindex(nodeDbId, reindexCallback); + Future future = this.reindexService.reindex(nodeDbId, reindexCallback); + + if (this.waitAfterSolrNodeActionMillis > 0L) { + Thread.sleep(this.waitAfterSolrNodeActionMillis); + } + + return future; } private String formatForFts(QName qname) { diff --git a/shared/src/main/resources/alfresco/module/com_inteligr8_alfresco_asie-shared/alfresco-global.properties b/shared/src/main/resources/alfresco/module/com_inteligr8_alfresco_asie-shared/alfresco-global.properties index f1248ab..df414f6 100644 --- a/shared/src/main/resources/alfresco/module/com_inteligr8_alfresco_asie-shared/alfresco-global.properties +++ b/shared/src/main/resources/alfresco/module/com_inteligr8_alfresco_asie-shared/alfresco-global.properties @@ -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