Compare commits

...
11 Commits
22 changed files with 285 additions and 259 deletions
+21 -20
View File
@@ -6,21 +6,16 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3.4</version>
<version>1.3.6</version>
<relativePath>../</relativePath>
</parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-api</artifactId>
<version>1.1.1-asie2</version>
<version>1.1.2-asie2</version>
<packaging>jar</packaging>
<name>ASIE Jakarta RS API</name>
<description>Alfresco Search &amp; Insight Engine Jakarta RS API</description>
<properties>
<alfresco.platform.version>23.2.0</alfresco.platform.version>
</properties>
<dependencyManagement>
<dependencies>
@@ -43,17 +38,6 @@
<dependency>
<groupId>org.alfresco</groupId>
<artifactId>alfresco-data-model</artifactId>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.17.0</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
@@ -63,7 +47,7 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>3.0.2-jersey</version>
<version>${commom-rest-client.base.version}-jersey</version>
<scope>test</scope>
</dependency>
<dependency>
@@ -75,10 +59,27 @@
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>5.11.2</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<extensions>true</extensions>
<configuration>
<tiles>
<!-- Documentation: https://git.inteligr8.com/inteligr8/ootbee-beedk/src/stable/beedk-acs-platform-webapp-tile -->
<!-- TODO spin up just ASIE
<tile>com.inteligr8.ootbee:beedk-acs-platform-webapp-tile:[1.1.6,2.0.0)</tile>
-->
</tiles>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
@@ -24,7 +24,7 @@ public class IndexRequest extends JsonFormattedResponseRequest<IndexRequest> {
@QueryParam("acltxid")
private Long aclTransactionId;
@QueryParam("nodeId")
@QueryParam("nodeid")
private Long nodeId;
@QueryParam("aclid")
@@ -3,7 +3,7 @@ package com.inteligr8.alfresco.asie.model.core;
import java.util.Collection;
import org.alfresco.service.cmr.repository.StoreRef;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.StringUtils;
import com.inteligr8.solr.model.JsonFormattedResponseRequest;
@@ -89,7 +89,7 @@ public class NewCoreRequest extends JsonFormattedResponseRequest<NewCoreRequest>
}
public NewCoreRequest withShardIds(Collection<String> shardIds) {
this.shardIds = StringUtils.join(shardIds, ",");
this.shardIds = StringUtils.collectionToDelimitedString(shardIds, ",");
return this;
}
@@ -24,7 +24,7 @@ public class PurgeRequest extends JsonFormattedResponseRequest<PurgeRequest> {
@QueryParam("acltxid")
private Long aclTransactionId;
@QueryParam("nodeId")
@QueryParam("nodeid")
private Long nodeId;
@QueryParam("aclid")
@@ -24,7 +24,7 @@ public class ReindexRequest extends JsonFormattedResponseRequest<ReindexRequest>
@QueryParam("acltxid")
private Long aclTransactionId;
@QueryParam("nodeId")
@QueryParam("nodeid")
private Long nodeId;
@QueryParam("aclid")
@@ -8,7 +8,7 @@ import org.slf4j.LoggerFactory;
import com.inteligr8.alfresco.asie.AsieClient;
public class AbstractApiUnitTest {
public class AbstractApiIT {
protected Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -11,7 +11,7 @@ import com.inteligr8.solr.model.Action.Status;
import com.inteligr8.solr.model.Cores;
import com.inteligr8.solr.model.ResponseHeader;
public class CoreAdminReindexUnitTest extends AbstractApiUnitTest {
public class CoreAdminReindexIT extends AbstractApiIT {
@Test
public void reindex() {
@@ -19,7 +19,7 @@ import com.inteligr8.solr.model.core.StatusResponse;
import jakarta.ws.rs.ProcessingException;
public class CoreAdminStatusUnitTest extends AbstractApiUnitTest {
public class CoreAdminStatusIT extends AbstractApiIT {
@Test
public void noHost() {
+12 -19
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3.4</version>
<version>1.3.6</version>
<relativePath>../</relativePath>
</parent>
@@ -16,10 +16,9 @@
<name>ASIE Platform Module for ACS Enterprise</name>
<properties>
<alfresco.sdk.version>4.9.0</alfresco.sdk.version>
<alfresco.platform.version>23.3.0</alfresco.platform.version>
<alfresco.platform.war.version>23.3.0.98</alfresco.platform.war.version>
<tomcat-rad.version>10-2.1</tomcat-rad.version>
<alfresco.platform.version>25.1.0</alfresco.platform.version>
<alfresco.platform.war.version>25.1.0.71</alfresco.platform.war.version>
<tomcat-rad.version>2.3-tomcat-10.1.52</tomcat-rad.version>
<beedk.rad.acs-search.enabled>true</beedk.rad.acs-search.enabled>
</properties>
@@ -37,10 +36,13 @@
</dependencyManagement>
<dependencies>
<!-- Alfresco Modules required to use this module -->
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-shared</artifactId>
<version>${project.version}</version>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.4.0-acs-v25.1</version>
<type>amp</type>
<scope>provided</scope>
</dependency>
<!-- Needed by this module, but provided by ACS -->
@@ -50,19 +52,11 @@
<scope>provided</scope>
</dependency>
<!-- Alfresco Modules required to use this module -->
<!-- core dependency, but dependencies of provided ones above should override -->
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.3.2-acs-v23.3</version>
<type>amp</type>
</dependency>
<!-- Provided by cxf-jaxrs-platform-module, but packaged due to shared -->
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<scope>provided</scope>
<artifactId>asie-shared</artifactId>
<version>${project.version}</version>
</dependency>
<!-- Provided by cxf-jaxrs-platform-module, but packaged due to solr-api -->
@@ -108,7 +102,6 @@
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.40</version>
<extensions>true</extensions>
<configuration>
<tiles>
+30 -19
View File
@@ -5,7 +5,7 @@
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3.4</version>
<version>1.3.6</version>
<packaging>pom</packaging>
<name>ASIE Platform Module Parent</name>
@@ -43,15 +43,21 @@
<maven.compiler.target>17</maven.compiler.target>
<maven.compiler.release>17</maven.compiler.release>
<maven.deploy.skip>true</maven.deploy.skip>
<!-- must be aligned with the target ACS platform junit -->
<junit.version>5.12.2</junit.version>
<commom-rest-client.base.version>3.0.3</commom-rest-client.base.version>
<alfresco.platform.version>25.1.0</alfresco.platform.version>
</properties>
<dependencyManagement>
<dependencies>
<!-- Provided by cxf-jaxrs-platform-module, but packaged due to shared -->
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>3.0.3-cxf</version>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>${junit.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
@@ -59,43 +65,48 @@
<build>
<pluginManagement>
<plugins>
<!-- avoids log4j dependency -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.14.1</version>
</plugin>
<!-- avoids struts dependency -->
<!-- helps avoid vulnerable dependencies -->
<plugin>
<artifactId>maven-site-plugin</artifactId>
<version>3.21.0</version>
</plugin>
<!-- Force use of a new maven-dependency-plugin that doesn't download struts dependency -->
<plugin>
<artifactId>maven-dependency-plugin</artifactId>
<version>3.9.0</version>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.15.0</version>
</plugin>
<plugin>
<artifactId>maven-dependency-plugin</artifactId>
<version>3.10.0</version>
</plugin>
<plugin>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.5.4</version>
<version>3.5.5</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.14.0</version>
<version>${junit.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<artifactId>maven-failsafe-plugin</artifactId>
<version>3.5.4</version>
<version>3.5.5</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.14.0</version>
<version>${junit.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.43</version>
</plugin>
</plugins>
</pluginManagement>
</build>
+4 -2
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3.4</version>
<version>1.3.6</version>
<relativePath>../</relativePath>
</parent>
@@ -17,7 +17,8 @@
<properties>
<alfresco.sdk.version>4.9.0</alfresco.sdk.version>
<alfresco.platform.version>23.3.0</alfresco.platform.version>
<common-rest-client.version>${commom-rest-client.base.version}-cxf</common-rest-client.version>
</properties>
<dependencyManagement>
@@ -41,6 +42,7 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>${common-rest-client.version}</version>
</dependency>
<!-- Needed by this module, but provided by ACS -->
@@ -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;
@@ -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() {
@@ -8,23 +8,22 @@ 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 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.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
@@ -41,7 +40,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());
@@ -61,10 +60,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() {
@@ -79,6 +78,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
@@ -133,16 +148,15 @@ public abstract class AbstractNodeActionService {
}
}
private Future<Void> _action(long nodeDbId, ActionCallback callback, Long fullQueueExpireTimeMillis) throws TimeoutException, InterruptedException {
private Future<Void> _action(
final long nodeDbId,
final ActionCallback callback,
Long fullQueueExpireTimeMillis) throws TimeoutException, InterruptedException {
List<com.inteligr8.alfresco.asie.model.ShardInstance> eligibleInstances = this.findPossibleShardInstances(nodeDbId);
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);
@@ -150,49 +164,12 @@ public abstract class AbstractNodeActionService {
Callable<Void> callable = new Callable<>() {
@Override
public Void call() {
String core = instance.extractShard().getCoreName();
SolrHost host = instance.extractNode();
URL url = host.toUrl(apiService.isSecure() ? "https" : "http");
CoreAdminApi api = apiService.createApi(url.toString(), CoreAdminApi.class);
try {
logger.debug("Performing {} of ACS node against shard instance: {}: {}", getActionName(), nodeDbId, instance);
BaseResponse apiResponse = execute(api, core, nodeDbId);
logger.trace("Performed {} of ACS node against shard instance: {}: {}", getActionName(), nodeDbId, instance);
Action action = null;
if (apiResponse instanceof ActionCoreResponse<?>) {
action = ((ActionCoreResponse<Action>) apiResponse).getCores().getByCore(core);
} else if (apiResponse instanceof ActionResponse<?>) {
action = ((ActionResponse<Action>) apiResponse).getAction();
}
if (action == null) {
callback.unknownResult(instance);
} else {
switch (action.getStatus()) {
case Scheduled:
callback.scheduled(instance);
break;
case Success:
callback.success(instance);
break;
default:
if (apiResponse instanceof com.inteligr8.alfresco.asie.model.BaseResponse) {
com.inteligr8.alfresco.asie.model.BaseResponse asieResponse = (com.inteligr8.alfresco.asie.model.BaseResponse) apiResponse;
logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}", getActionName(), nodeDbId, instance, asieResponse.getException());
callback.error(instance, asieResponse.getException());
} else {
logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}", getActionName(), nodeDbId, instance, apiResponse.getResponseHeader().getStatus());
callback.error(instance, String.valueOf(apiResponse.getResponseHeader().getStatus()));
}
}
}
actionToShard(nodeDbId, callback, instance);
} catch (Exception e) {
logger.error("An exception occurred", e);
logger.error("An unexpected exception occurred", e);
callback.error(instance, e.getMessage());
}
return null;
}
};
@@ -207,15 +184,58 @@ public abstract class AbstractNodeActionService {
return future;
}
@SuppressWarnings("unchecked")
protected void actionToShard(long nodeDbId, ActionCallback callback, com.inteligr8.alfresco.asie.model.ShardInstance instance) {
String core = instance.extractShard().getCoreName();
SolrHost host = instance.extractNode();
URL url = host.toUrl(this.apiService.isSecure() ? "https" : "http");
CoreAdminApi api = this.apiService.createApi(url.toString(), CoreAdminApi.class);
this.logger.debug("Performing {} of ACS node against shard instance: {}: {}", this.getActionName(), nodeDbId, instance);
BaseResponse apiResponse = execute(api, core, nodeDbId);
this.logger.trace("Performed {} of ACS node against shard instance: {}: {}", this.getActionName(), nodeDbId, instance);
Action action = null;
if (apiResponse instanceof ActionCoreResponse<?>) {
action = ((ActionCoreResponse<Action>) apiResponse).getCores().getByCore(core);
} else if (apiResponse instanceof ActionResponse<?>) {
action = ((ActionResponse<Action>) apiResponse).getAction();
}
if (action == null) {
callback.unknownResult(instance);
} else {
switch (action.getStatus()) {
case Scheduled:
callback.scheduled(instance);
break;
case Success:
callback.success(instance);
break;
default:
if (apiResponse instanceof com.inteligr8.alfresco.asie.model.BaseResponse) {
com.inteligr8.alfresco.asie.model.BaseResponse asieResponse = (com.inteligr8.alfresco.asie.model.BaseResponse) apiResponse;
this.logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}",
this.getActionName(), nodeDbId, instance, asieResponse.getException());
callback.error(instance, asieResponse.getException());
} else {
this.logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}",
this.getActionName(), nodeDbId, instance, apiResponse.getResponseHeader().getStatus());
callback.error(instance, String.valueOf(apiResponse.getResponseHeader().getStatus()));
}
}
}
}
protected abstract BaseResponse execute(CoreAdminApi api, String core, long nodeDbId);
private List<com.inteligr8.alfresco.asie.model.ShardInstance> 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);
// SearchParameters searchParams = new SearchParameters();
// searchParams.setLanguage(SearchService.LANGUAGE_FTS_ALFRESCO);
// searchParams.setQuery("@" + this.formatForFts(ContentModel.PROP_NODE_DBID) + ":" + nodeDbId);
List<com.inteligr8.alfresco.asie.model.ShardInstance> instances = new LinkedList<>();
@@ -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) {
@@ -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
@@ -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);
}
}
@@ -1,6 +1,9 @@
package com.inteligr8.alfresco.asie.service;
import java.nio.charset.Charset;
import java.time.Instant;
import java.time.temporal.TemporalAccessor;
import java.util.Date;
import java.util.regex.Matcher;
import org.alfresco.error.AlfrescoRuntimeException;
@@ -9,6 +12,7 @@ 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.joda.time.ReadablePartial;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -70,7 +74,20 @@ public class SolrShardHashService {
}
this.logger.trace("Discovered node property for sharding: {} => {}", nodeRef, fullPropertyValue);
String hashableValue = fullPropertyValue.toString();
String hashableValue = null;
if (fullPropertyValue instanceof TemporalAccessor) {
Instant instant = Instant.from((TemporalAccessor) fullPropertyValue);
hashableValue = instant.toString();
} else if (fullPropertyValue instanceof ReadablePartial) {
hashableValue = ((ReadablePartial) fullPropertyValue).toString();
} else if (fullPropertyValue instanceof Date) {
Instant instant = ((Date) fullPropertyValue).toInstant();
hashableValue = instant.toString();
} else {
hashableValue = fullPropertyValue.toString();
}
if (shardset.getRegex() != null) {
Matcher matcher = shardset.getRegex().matcher(hashableValue);
if (!matcher.find()) {
@@ -84,6 +101,7 @@ public class SolrShardHashService {
int shardHash = this.hash(hashableValue, shardset.getShards().intValue());
this.logger.debug("Hash shardable value to shard instance ID: {}: {} => {}", nodeRef, hashableValue, shardHash);
return shardHash;
// TODO replicate hash algorithm for other shard methods
default:
this.logger.trace("Unable to determine shard instance ID due to shard method: {}", shardset.getMethod());
return -1;
@@ -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);
@@ -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;
@@ -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
+2 -2
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3.4</version>
<version>1.3.6</version>
<relativePath>../</relativePath>
</parent>
@@ -18,7 +18,7 @@
<name>Apache Solr Jakarta RS API</name>
<properties>
<jackson.version>2.18.0</jackson.version>
<jackson.version>2.21.1</jackson.version>
</properties>
<dependencies>