From 2c3fc8495d20c2b1fee5603ea279733d59b8505f Mon Sep 17 00:00:00 2001 From: "Brian M. Long" Date: Mon, 12 Jan 2026 11:51:28 -0500 Subject: [PATCH] add index support to reconcile --- .../asie/rest/ReconcileAcsNodesWebScript.java | 3 +- .../asie/service/AcsReconcileService.java | 109 ++++++++++++++---- .../asie/reconcileAcsNodes.post.desc.xml | 9 +- 3 files changed, 94 insertions(+), 27 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 a7e5a46..f2b7c96 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 @@ -32,6 +32,7 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript { public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException { final int fromDbId = this.getRequestTemplateIntegerVariable(request, "fromDbId"); final int toDbId = this.getRequestTemplateIntegerVariable(request, "toDbId"); + final boolean index = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "index", Boolean.class)); final boolean reindex = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "reindex", Boolean.class)); final boolean includeReconciled = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "includeReconciled", Boolean.class)); @@ -101,7 +102,7 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript { }; try { - this.reconcileService.reconcile(fromDbId, toDbId, null, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES); + this.reconcileService.reconcile(fromDbId, toDbId, null, index, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES); if (responseMap.containsKey("error")) { response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value()); 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 048f373..fba1e86 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 @@ -32,6 +32,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import com.inteligr8.alfresco.asie.model.ShardInstance; +import com.inteligr8.alfresco.asie.spi.IndexCallback; import com.inteligr8.alfresco.asie.spi.ReconcileCallback; import com.inteligr8.alfresco.asie.spi.ReindexCallback; import com.inteligr8.alfresco.asie.util.CompositeFuture; @@ -55,6 +56,9 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @Autowired private SearchService searchService; + @Autowired + private IndexService indexService; + @Autowired private ReindexService reindexService; @@ -109,20 +113,21 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { */ public void reconcile( long fromDbId, long toDbId, Integer nodesChunkSize, - boolean reindexUnreconciled, + boolean indexUnreconciled, + boolean reindexReconciled, 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, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout)); + 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, reindexUnreconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit)); + future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit)); future.purge(true); } @@ -135,18 +140,19 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { public Future reconcile( long fromDbId, long toDbId, Integer nodesChunkSize, - boolean reindexUnreconciled, + boolean indexUnreconciled, + boolean reindexReconciled, ReconcileCallback callback) throws InterruptedException { if (nodesChunkSize == null) nodesChunkSize = this.nodesChunkSize; - this.logger.trace("reconcile({}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled); + 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, reindexUnreconciled, callback, -1L, null, -1L, null)); + future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null)); future.purge(true); } } catch (TimeoutException te) { @@ -158,12 +164,13 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { protected Future reconcileChunk( long fromDbId, long toDbId, - boolean reindexUnreconciled, + boolean indexUnreconciled, + boolean reindexReconciled, ReconcileCallback callback, long queueTimeout, TimeUnit queueUnit, long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException { if (this.logger.isTraceEnabled()) - this.logger.trace("reconcileChunk({}, {}, {}, {}, {})", fromDbId, toDbId, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout)); + this.logger.trace("reconcileChunk({}, {}, {}, {}, {}, {})", fromDbId, toDbId, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout)); int dbIdCount = (int) (toDbId - fromDbId); @@ -202,22 +209,32 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { final long nodeDbId = _nodeDbId; this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId); + Callable callable; + final int dbIdIndex = (int) (nodeDbId - fromDbId); if (nodeRefs[dbIdIndex] != null) { - this.logger.trace("A node in the DB is already indexed in Solr: {}: {}", nodeDbId, nodeRefs[dbIdIndex]); - this.reconcileLogger.info("RECONCILED: {} <=> {}", nodeDbId, nodeRefs[dbIdIndex]); - callback.reconciled(nodeDbId); - continue; + callable = new Callable() { + @Override + public Void call() throws InterruptedException, TimeoutException { + logger.trace("A node in the DB is already indexed in Solr: {}: {}", nodeDbId, nodeRefs[dbIdIndex]); + reconcileLogger.info("RECONCILED: {} <=> {}", nodeDbId, nodeRefs[dbIdIndex]); + callback.reconciled(nodeDbId); + + if (reindexReconciled) + reindex(nodeDbId, nodeRefs[dbIdIndex], callback, execTimeout, execUnit); + return null; + } + }; + } else { + callable = new Callable() { + @Override + public Void call() throws InterruptedException, TimeoutException { + reconcile(nodeDbId, indexUnreconciled, callback, execTimeout, execUnit); + return null; + } + }; } - Callable callable = new Callable() { - @Override - public Void call() throws InterruptedException, TimeoutException { - reconcile(nodeDbId, reindexUnreconciled, callback, execTimeout, execUnit); - return null; - } - }; - if (queueTimeout < 0L) { future.combine(this.executor.submit(callable, -1L, null)); } else { @@ -229,7 +246,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { } public void reconcile(long nodeDbId, - boolean reindexUnreconciled, + boolean index, ReconcileCallback callback, long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException { NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId); @@ -250,16 +267,58 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { return; } - if (!reindexUnreconciled) { + 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); } else { - logger.debug("A node in the DB is not indexed in Solr; attempt to reindex: {}: {}", nodeDbId, nodeRef); - this.reindex(nodeDbId, nodeRef, callback, execTimeout, execUnit); + logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef); + this.index(nodeDbId, nodeRef, callback, execTimeout, execUnit); } } + 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<>(); + + IndexCallback indexCallback = new IndexCallback() { + + @Override + public void success(ShardInstance instance) { + reconcileLogger.info("INDEXED: {} <=> {}", nodeDbId, nodeRef); + syncHosts.add(instance); + } + + @Override + public void scheduled(ShardInstance instance) { + reconcileLogger.info("INDEXING: {} <=> {}", nodeDbId, nodeRef); + asyncHosts.add(instance); + } + + @Override + public void error(ShardInstance instance, String message) { + reconcileLogger.info("FAILED INDEX: {} <=> {}", nodeDbId, nodeRef); + errorHosts.put(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); + } + public void reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback, long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException { @@ -283,7 +342,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @Override public void error(ShardInstance instance, String message) { - reconcileLogger.info("UNINDEXED: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("FAILED REINDEX: {} <=> {}", nodeDbId, nodeRef); errorHosts.put(instance, message); } }; diff --git a/shared/src/main/resources/alfresco/extension/templates/webscripts/com/inteligr8/alfresco/asie/reconcileAcsNodes.post.desc.xml b/shared/src/main/resources/alfresco/extension/templates/webscripts/com/inteligr8/alfresco/asie/reconcileAcsNodes.post.desc.xml index 82e3963..8f67475 100644 --- a/shared/src/main/resources/alfresco/extension/templates/webscripts/com/inteligr8/alfresco/asie/reconcileAcsNodes.post.desc.xml +++ b/shared/src/main/resources/alfresco/extension/templates/webscripts/com/inteligr8/alfresco/asie/reconcileAcsNodes.post.desc.xml @@ -16,6 +16,13 @@
toDbId
A DB ID integer for the ending point of a range, exclusive.
+

The following query parameters are also supported:

+
+
index
+
Schedule an `INDEX` action against all unreconciled nodes.
+
reindex
+
Schedule a `REINDEX` action against all reconciled nodes.
+

The following response body should be expected in most cases (200, 202, and 500 status codes):

 			{
@@ -58,7 +65,7 @@
 	]]>
 	
 	
-	/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?reindex={reindex?}
+	/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?index={index?}&reindex={reindex?}
 	any