diff --git a/enterprise-module/pom.xml b/enterprise-module/pom.xml index ca77f5d..db5079f 100644 --- a/enterprise-module/pom.xml +++ b/enterprise-module/pom.xml @@ -69,6 +69,7 @@ jakarta.ws.rs jakarta.ws.rs-api + 4.0.0 provided @@ -79,11 +80,13 @@ com.fasterxml.jackson.datatype jackson-datatype-jsr310 + 2.17.3 provided com.fasterxml.jackson.module jackson-module-jakarta-xmlbind-annotations + 2.17.2 provided 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 8beadc7..5b42d0e 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 @@ -11,6 +11,7 @@ import java.util.concurrent.ExecutionException; 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; @@ -18,6 +19,8 @@ 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; @@ -48,12 +51,18 @@ public abstract class AbstractNodeActionService { @Autowired private NamespaceService namespaceService; + @Autowired + private NodeService nodeService; + @Autowired private ApiService apiService; @Autowired private ExecutorManager executorManager; + @Autowired + private SolrShardHashService shardHashService; + @Autowired(required = false) @Qualifier(Constants.QUALIFIER_ASIE) private ShardRegistry shardRegistry; @@ -216,42 +225,56 @@ public abstract class AbstractNodeActionService { List instances = new LinkedList<>(); - List slicedInstances = this.shardRegistry.getIndexSlice(searchParams); - if (slicedInstances != null) { - this.logger.trace("Due to a sharding method, considering only applicable shards and their ASIE nodes: {}: {}", nodeDbId, slicedInstances); - - for (ShardInstance instance : slicedInstances) - instances.add(this.toModel(instance)); - } else { - for (Entry>> floc : this.shardRegistry.getFlocs().entrySet()) { - if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE)) - continue; - for (Entry> shard : floc.getValue().entrySet()) { - for (ShardState shardState : shard.getValue()) + // we need a ShardRegistry method like getIndexSlice, but + // (1) works with shard methods other than explicit + // (2) returns all possible instances/nodes, not just one node per instance + + for (Entry>> floc : this.shardRegistry.getFlocs().entrySet()) { + if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE)) + continue; + + Integer shardHash = null; + ShardSet shardset = null; + for (Entry> shard : floc.getValue().entrySet()) { + for (ShardState shardState : shard.getValue()) { + // every shard and instance (state) in the FLOC is the same + // but we need one ShardState to determine the full configuration + // 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); + } + } + + if (shardHash != null) { + if (shard.getKey().getInstance() == shardHash) + instances.add(this.toModel(shardState.getShardInstance(), shardState)); + } else { instances.add(this.toModel(shardState.getShardInstance(), shardState)); + } } } - - this.logger.trace("Despite sharding, considering all shards and nodes: {}: {}", nodeDbId, instances); } + return instances; } - private com.inteligr8.alfresco.asie.model.ShardInstance toModel(ShardInstance instance) { - // get any random shardState - Floc floc = instance.getShard().getFloc(); - Map> shardsStates = this.shardRegistry.getFlocs().get(floc); - if (shardsStates == null) - throw new IllegalStateException(); - Set shardStates = shardsStates.get(instance.getShard()); - if (shardStates == null || shardStates.isEmpty()) - throw new IllegalStateException(); - ShardState anyShardState = shardStates.iterator().next(); - - return this.toModel(instance, anyShardState); - } - private com.inteligr8.alfresco.asie.model.ShardInstance toModel(ShardInstance instance, ShardState anyShardState) { Floc floc = instance.getShard().getFloc(); 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 e494a85..1f38868 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 @@ -289,19 +289,19 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @Override public void success(ShardInstance instance) { - reconcileLogger.info("INDEXED: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("INDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance); syncHosts.add(instance); } @Override public void scheduled(ShardInstance instance) { - reconcileLogger.info("INDEXING: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("INDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance); asyncHosts.add(instance); } @Override public void error(ShardInstance instance, String message) { - reconcileLogger.info("FAILED INDEX: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("FAILED INDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance); errorHosts.put(instance, message); } }; @@ -331,19 +331,19 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { @Override public void success(ShardInstance instance) { - reconcileLogger.info("REINDEXED: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("REINDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance); syncHosts.add(instance); } @Override public void scheduled(ShardInstance instance) { - reconcileLogger.info("REINDEXING: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("REINDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance); asyncHosts.add(instance); } @Override public void error(ShardInstance instance, String message) { - reconcileLogger.info("FAILED REINDEX: {} <=> {}", nodeDbId, nodeRef); + reconcileLogger.info("FAILED REINDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance); errorHosts.put(instance, message); } };