Compare commits

..
16 Commits
Author SHA1 Message Date
brian.long fe2eaa0588 v1.1.0 model poms 2025-03-05 16:21:14 -05:00
brian.long 68471be9ef Merge branch 'develop' into stable 2025-03-05 16:18:49 -05:00
brian.long 82410805db v1.2.2 poms 2025-01-09 16:08:48 -05:00
brian.long ceb8d2c690 Merge branch 'develop' into stable 2025-01-09 16:05:18 -05:00
brian.long 1c15853652 v1.2.1 poms 2025-01-08 11:56:21 -05:00
brian.long cf1bc8d2be Merge branch 'develop' into stable 2025-01-08 11:55:31 -05:00
brian.long a49ddbd373 Merge branch 'develop' into stable 2024-11-13 18:05:16 -05:00
brian.long 738e5b98f8 v1.1.1 poms 2024-11-06 13:26:28 -05:00
brian.long 9d9d17c49b Merge branch 'develop' into stable 2024-11-06 13:25:38 -05:00
brian.long 43109e7d9c Merge branch 'develop' into stable 2024-10-31 15:17:54 -04:00
brian.long dcb37c3dc8 v1.1.0 poms 2024-10-31 14:53:06 -04:00
brian.long 68ffb56cc9 Merge branch 'develop' into stable 2024-10-31 14:50:33 -04:00
brian.long dc436e9ce0 v1.0.1 poms 2024-10-29 09:29:14 -04:00
brian.long 18c39b0fc5 fix merge 2024-10-29 09:28:21 -04:00
brian.long b7aa615103 Merge branch 'develop' into stable 2024-10-29 09:27:47 -04:00
brian.long e8f133b873 v1.0.0 poms 2024-10-28 16:26:00 -04:00
34 changed files with 373 additions and 875 deletions
+23 -24
View File
@@ -6,16 +6,21 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-api</artifactId>
<version>1.2-SNAPSHOT-asie2</version>
<version>1.1.0-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>
@@ -33,11 +38,22 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>solr-api</artifactId>
<version>1.2-SNAPSHOT-solr6</version>
<version>1.1.0-solr6</version>
</dependency>
<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>
@@ -47,44 +63,27 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>${commom-rest-client.base.version}-jersey</version>
<version>3.0.2-jersey</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.glassfish.jersey.inject</groupId>
<artifactId>jersey-hk2</artifactId>
<version>4.0.2</version>
<version>3.1.10</version>
<scope>test</scope>
</dependency>
<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>
<id>alfresco-public</id>
<url>https://artifacts.alfresco.com/nexus/repository/releases/</url>
<url>https://artifacts.alfresco.com/nexus/repository/public/</url>
</repository>
</repositories>
</project>
@@ -7,7 +7,6 @@ import com.inteligr8.alfresco.asie.model.core.DisableIndexingRequest;
import com.inteligr8.alfresco.asie.model.core.EnableIndexingRequest;
import com.inteligr8.alfresco.asie.model.core.FixAction;
import com.inteligr8.alfresco.asie.model.core.FixRequest;
import com.inteligr8.alfresco.asie.model.core.IndexRequest;
import com.inteligr8.alfresco.asie.model.core.IndexingStatusAction;
import com.inteligr8.alfresco.asie.model.core.NewCoreRequest;
import com.inteligr8.alfresco.asie.model.core.NewDefaultIndexRequest;
@@ -62,10 +61,6 @@ public interface CoreAdminApi extends com.inteligr8.solr.api.CoreAdminApi {
@Produces(MediaType.APPLICATION_JSON)
ActionCoreResponse<Action> purge(@BeanParam PurgeRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionCoreResponse<Action> index(@BeanParam IndexRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionCoreResponse<Action> reindex(@BeanParam ReindexRequest request);
@@ -1,106 +0,0 @@
package com.inteligr8.alfresco.asie.model.core;
import com.inteligr8.solr.model.JsonFormattedResponseRequest;
import jakarta.annotation.Nonnull;
import jakarta.ws.rs.DefaultValue;
import jakarta.ws.rs.QueryParam;
public class IndexRequest extends JsonFormattedResponseRequest<IndexRequest> {
private static final String ACTION = "index";
@QueryParam("action")
@DefaultValue(ACTION)
@Nonnull
private String action = ACTION;
@QueryParam("core")
private String core;
@QueryParam("txid")
private Long transactionId;
@QueryParam("acltxid")
private Long aclTransactionId;
@QueryParam("nodeid")
private Long nodeId;
@QueryParam("aclid")
private Long aclId;
public String getAction() {
return action;
}
public void setAction(String action) {
this.action = action;
}
public String getCore() {
return core;
}
public void setCore(String core) {
this.core = core;
}
public IndexRequest withCore(String core) {
this.core = core;
return this;
}
public Long getTransactionId() {
return transactionId;
}
public void setTransactionId(Long transactionId) {
this.transactionId = transactionId;
}
public IndexRequest withTransactionId(Long transactionId) {
this.transactionId = transactionId;
return this;
}
public Long getAclTransactionId() {
return aclTransactionId;
}
public void setAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
}
public IndexRequest withAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
return this;
}
public Long getNodeId() {
return nodeId;
}
public void setNodeId(Long nodeId) {
this.nodeId = nodeId;
}
public IndexRequest withNodeId(Long nodeId) {
this.nodeId = nodeId;
return this;
}
public Long getAclId() {
return aclId;
}
public void setAclId(Long aclId) {
this.aclId = aclId;
}
public IndexRequest withAclId(Long aclId) {
this.aclId = aclId;
return this;
}
}
@@ -3,7 +3,7 @@ package com.inteligr8.alfresco.asie.model.core;
import java.util.Collection;
import org.alfresco.service.cmr.repository.StoreRef;
import org.springframework.util.StringUtils;
import org.apache.commons.lang3.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.collectionToDelimitedString(shardIds, ",");
this.shardIds = StringUtils.join(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 AbstractApiIT {
public class AbstractApiUnitTest {
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 CoreAdminReindexIT extends AbstractApiIT {
public class CoreAdminReindexUnitTest extends AbstractApiUnitTest {
@Test
public void reindex() {
@@ -19,7 +19,7 @@ import com.inteligr8.solr.model.core.StatusResponse;
import jakarta.ws.rs.ProcessingException;
public class CoreAdminStatusIT extends AbstractApiIT {
public class CoreAdminStatusUnitTest extends AbstractApiUnitTest {
@Test
public void noHost() {
+8 -7
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
@@ -16,10 +16,10 @@
<name>ASIE Platform Module for ACS Community</name>
<properties>
<alfresco.sdk.version>4.16.0</alfresco.sdk.version>
<alfresco.platform.version>26.1.0</alfresco.platform.version>
<alfresco.platform.war.version>26.1.0.61</alfresco.platform.war.version>
<tomcat-rad.version>2.3-tomcat-11.0.22</tomcat-rad.version>
<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>
<beedk.rad.acs-search.enabled>true</beedk.rad.acs-search.enabled>
</properties>
@@ -59,7 +59,7 @@
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.4.0-acs-v26.1</version>
<version>1.3.1-acs-v23.3</version>
<type>amp</type>
</dependency>
@@ -81,6 +81,7 @@
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.40</version>
<extensions>true</extensions>
<configuration>
<tiles>
@@ -99,7 +100,7 @@
<repositories>
<repository>
<id>alfresco-public</id>
<url>https://artifacts.alfresco.com/nexus/repository/releases/</url>
<url>https://artifacts.alfresco.com/nexus/content/groups/public</url>
</repository>
</repositories>
</project>
+41 -38
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
@@ -16,10 +16,10 @@
<name>ASIE Platform Module for ACS Enterprise</name>
<properties>
<alfresco.platform.version>26.1.0</alfresco.platform.version>
<alfresco.platform.war.version>26.1.0.61</alfresco.platform.war.version>
<tomcat-rad.version>2.3-tomcat-11.0.22</tomcat-rad.version>
<jackson.version>2.17.2</jackson.version>
<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>
<beedk.rad.acs-search.enabled>true</beedk.rad.acs-search.enabled>
</properties>
@@ -33,17 +33,43 @@
<type>pom</type>
<scope>import</scope>
</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>
</dependency>
<!-- Provided by cxf-jaxrs-platform-module, but packaged due to solr-api -->
<dependency>
<groupId>jakarta.ws.rs</groupId>
<artifactId>jakarta.ws.rs-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>jakarta.annotation</groupId>
<artifactId>jakarta.annotation-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-jakarta-xmlbind-annotations</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<!-- Alfresco Modules required to use this module -->
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.4.0-acs-v26.1</version>
<type>amp</type>
<scope>provided</scope>
<artifactId>asie-shared</artifactId>
<version>${project.version}</version>
</dependency>
<!-- Needed by this module, but provided by ACS -->
@@ -53,36 +79,12 @@
<scope>provided</scope>
</dependency>
<!-- core dependency, but dependencies of provided ones above should override -->
<!-- Alfresco Modules required to use this module -->
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-shared</artifactId>
<version>${project.version}</version>
</dependency>
<!-- Provided by cxf-jaxrs-platform-module, but packaged due to solr-api -->
<dependency>
<groupId>jakarta.ws.rs</groupId>
<artifactId>jakarta.ws.rs-api</artifactId>
<version>4.0.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>jakarta.annotation</groupId>
<artifactId>jakarta.annotation-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>${jackson.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-jakarta-xmlbind-annotations</artifactId>
<version>${jackson.version}</version>
<scope>provided</scope>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.3.2-acs-v23.3</version>
<type>amp</type>
</dependency>
<!-- Including for testing purposes only -->
@@ -103,6 +105,7 @@
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.40</version>
<extensions>true</extensions>
<configuration>
<tiles>
@@ -1,33 +1,21 @@
package com.inteligr8.alfresco.asie.enterprise.rest;
import org.alfresco.repo.index.shard.ShardState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.extensions.webscripts.WebScriptException;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.model.ShardSet;
import com.inteligr8.alfresco.asie.rest.model.NodeShardParameterSet;
import com.inteligr8.alfresco.asie.spi.ShardDiscoveryService;
@Component(value = "webscript.com.inteligr8.alfresco.asie.nodeShard.delete")
public class UnloadNodeShardWebScript extends AbstractUnregisterNodeWebScript<NodeShardParameterSet> {
@Autowired
private ShardDiscoveryService sds;
@Override
protected NodeShardParameterSet createParameters(WebScriptRequest req, String nodeHostname, int nodePort) {
String coreName = this.getRequiredPathParameter(req, "shardCore");
ShardSet shardSet = this.getRequiredPathParameter(req, "shardSet", ShardSet.class);
int shardId = this.getRequiredPathParameter(req, "shardId", Integer.class);
try {
ShardSet shardSet = this.sds.findSetByCore(coreName);
return new NodeShardParameterSet(nodeHostname, nodePort, shardSet, shardId);
} catch (IllegalArgumentException iae) {
throw new WebScriptException(HttpStatus.BAD_REQUEST.value(), iae.getMessage());
}
return new NodeShardParameterSet(nodeHostname, nodePort, shardSet, shardId);
}
@Override
@@ -50,15 +50,12 @@ public class ShardDiscoveryService implements com.inteligr8.alfresco.asie.spi.Sh
for (Entry<Shard, Set<ShardState>> flocShard : floc.getValue().entrySet()) {
for (ShardState shardState : flocShard.getValue()) {
ShardSet shardSet = ShardSet.from(floc.getKey(), shardState);
if (shardSet.getCore().equals(core)) {
this.logger.debug("Found matching shard set: {}: {}", core, shardSet);
if (shardSet.getCore().equals(core))
return shardSet;
}
}
}
}
this.logger.trace("Did not find matching shard set: {}", core);
return null;
}
@@ -99,8 +96,7 @@ public class ShardDiscoveryService implements com.inteligr8.alfresco.asie.spi.Sh
}
}
}
this.logger.trace("Did not find matching node: {}:{}", nodeHostname, nodePort);
return null;
}
@@ -131,7 +127,6 @@ public class ShardDiscoveryService implements com.inteligr8.alfresco.asie.spi.Sh
ShardInstance shardInstance = shardState.getShardInstance();
if (node.equals(SolrHost.from(shardInstance))) {
this.logger.trace("Found shard on node: {}: {}", node, shardInstance.getShard().getInstance());
Map<Integer, ShardInstanceState> shardStates = setShardStates.get(shardSet);
if (shardStates == null)
setShardStates.put(shardSet, shardStates = new HashMap<>());
@@ -157,9 +152,7 @@ public class ShardDiscoveryService implements com.inteligr8.alfresco.asie.spi.Sh
for (Entry<Floc, Map<Shard, Set<ShardState>>> floc : flocs.entrySet()) {
if (shardMethodSet.contains(floc.getKey().getShardMethod())) {
this.logger.trace("Found qualifying shard method: {}", floc.getKey().getShardMethod());
ShardState shardState = this.extractAnyShardState(floc.getValue());
this.logger.trace("Found single shard state representing shard method: {}: {}", floc.getKey().getShardMethod(), shardState);
shardSets.add(ShardSet.from(floc.getKey(), shardState));
}
}
@@ -264,7 +257,7 @@ public class ShardDiscoveryService implements com.inteligr8.alfresco.asie.spi.Sh
ShardInstanceState nodeShardState = ShardInstanceState.from(shardState);
Pair<SolrHost, ShardInstanceState> pair = new Pair<>(node, nodeShardState);
if (!shardNodeStates.containsKey(shardId) || comparator.compare(pair, shardNodeStates.get(shardId)) < 0)
if (comparator.compare(pair, shardNodeStates.get(shardId)) < 0)
shardNodeStates.put(shardId, pair);
}
}
+23 -45
View File
@@ -5,7 +5,7 @@
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<packaging>pom</packaging>
<name>ASIE Platform Module Parent</name>
@@ -43,70 +43,48 @@
<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.4</commom-rest-client.base.version>
<alfresco.platform.version>26.1.0</alfresco.platform.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>${junit.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<pluginManagement>
<plugins>
<!-- helps avoid vulnerable dependencies -->
<plugin>
<artifactId>maven-site-plugin</artifactId>
<version>3.22.0</version>
</plugin>
<!-- avoids log4j dependency -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.15.0</version>
<version>3.13.0</version>
</plugin>
<!-- avoids struts dependency -->
<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.11.0</version>
<version>3.8.1</version>
</plugin>
<plugin>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.5.6</version>
<version>3.4.0</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>${junit.version}</version>
<version>5.11.0</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<artifactId>maven-failsafe-plugin</artifactId>
<version>3.5.6</version>
<version>3.4.0</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>${junit.version}</version>
<version>5.11.0</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.45</version>
</plugin>
</plugins>
</pluginManagement>
</build>
@@ -120,7 +98,7 @@
<profiles>
<profile>
<id>central-publish</id>
<id>ossrh-release</id>
<properties>
<maven.deploy.skip>true</maven.deploy.skip>
</properties>
@@ -160,20 +138,20 @@
</executions>
</plugin>
<plugin>
<groupId>org.sonatype.central</groupId>
<artifactId>central-publishing-maven-plugin</artifactId>
<version>0.10.0</version>
<groupId>org.sonatype.plugins</groupId>
<artifactId>nexus-staging-maven-plugin</artifactId>
<version>1.7.0</version>
<extensions>true</extensions>
<configuration>
<publishingServerId>central</publishingServerId>
<autoPublish>true</autoPublish>
<serverId>ossrh</serverId>
<nexusUrl>https://s01.oss.sonatype.org/</nexusUrl>
<autoReleaseAfterClose>true</autoReleaseAfterClose>
</configuration>
<!-- for some reason this is required... -->
<executions>
<execution>
<id>deploy</id>
<id>ossrh-deploy</id>
<phase>deploy</phase>
<goals><goal>publish</goal></goals>
<goals><goal>deploy</goal></goals>
</execution>
</executions>
</plugin>
+6 -7
View File
@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
@@ -16,9 +16,8 @@
<name>ASIE Shared Library for Platform Modules</name>
<properties>
<alfresco.sdk.version>4.16.0</alfresco.sdk.version>
<common-rest-client.version>${commom-rest-client.base.version}-cxf</common-rest-client.version>
<alfresco.sdk.version>4.9.0</alfresco.sdk.version>
<alfresco.platform.version>23.3.0</alfresco.platform.version>
</properties>
<dependencyManagement>
@@ -37,12 +36,12 @@
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-api</artifactId>
<version>1.2-SNAPSHOT-asie2</version>
<version>1.0.0-asie2</version>
</dependency>
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>${common-rest-client.version}</version>
<version>3.0.3-cxf</version>
</dependency>
<!-- Needed by this module, but provided by ACS -->
@@ -73,7 +72,7 @@
<repositories>
<repository>
<id>alfresco-public</id>
<url>https://artifacts.alfresco.com/nexus/repository/releases/</url>
<url>https://artifacts.alfresco.com/nexus/content/groups/public</url>
</repository>
</repositories>
</project>
@@ -9,8 +9,14 @@ 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;
@@ -7,6 +7,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
import com.inteligr8.alfresco.asie.model.ShardSet;
public abstract class AbstractAsieNodeShardWebScript extends AbstractAsieShardableWebScript {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -20,14 +22,14 @@ public abstract class AbstractAsieNodeShardWebScript extends AbstractAsieShardab
String nodeHostname = colon < 0 ? nodeEndpoint : nodeEndpoint.substring(0, colon);
int nodePort = colon < 0 ? this.getApiService().getDefaultSolrPort() : Integer.parseInt(nodeEndpoint.substring(colon+1));
String shardCore = this.getRequiredPathParameter(req, "shardCore", String.class);
ShardSet shardSet = this.getRequiredPathParameter(req, "shardSet", ShardSet.class);
int shardId = this.getRequiredPathParameter(req, "shardId", Integer.class);
this.execute(req, res, nodeHostname, nodePort, shardCore, shardId);
this.execute(req, res, nodeHostname, nodePort, shardSet, shardId);
}
protected abstract void execute(WebScriptRequest req, WebScriptResponse res,
String nodeHostname, int nodePort, String shardCore, int shardId)
String nodeHostname, int nodePort, ShardSet shardSet, int shardId)
throws IOException;
}
@@ -1,24 +0,0 @@
package com.inteligr8.alfresco.asie.rest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.service.IndexService;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.indexAcsNode.post")
public class IndexAcsNodeWebScript extends AbstractAcsNodeActionWebScript {
@Autowired
private IndexService indexSerivce;
@Override
protected void executeAction(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
this.indexSerivce.index(nodeDbId, callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
}
}
@@ -32,7 +32,6 @@ 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));
@@ -44,10 +43,10 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript {
public void reconciled(long nodeDbId) {
if (includeReconciled) {
@SuppressWarnings("unchecked")
List<Long> reconciledNodeDbIds = (List<Long>) responseMap.get("reconciled");
if (reconciledNodeDbIds == null)
responseMap.put("reconciled", reconciledNodeDbIds = new LinkedList<>());
reconciledNodeDbIds.add(nodeDbId);
List<Long> unreconciledNodeDbIds = (List<Long>) responseMap.get("reconciled");
if (unreconciledNodeDbIds == null)
responseMap.put("reconciled", unreconciledNodeDbIds = new LinkedList<>());
unreconciledNodeDbIds.add(nodeDbId);
}
}
@@ -102,7 +101,7 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript {
};
try {
this.reconcileService.reconcile(fromDbId, toDbId, null, index, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES);
this.reconcileService.reconcile(fromDbId, toDbId, null, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES);
if (responseMap.containsKey("error")) {
response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
@@ -12,12 +12,17 @@ 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;
@@ -40,6 +45,9 @@ public abstract class AbstractActionService {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private NamespaceService namespaceService;
@Autowired
private ApiService apiService;
@@ -50,10 +58,10 @@ public abstract class AbstractActionService {
@Qualifier(Constants.QUALIFIER_ASIE)
private ShardRegistry shardRegistry;
@Value("${inteligr8.asie.default.concurrentQueueSize}")
@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.default.concurrency}")
@Value("${inteligr8.asie.default.concurrency:16}")
private int concurrency;
protected int getConcurrency() {
@@ -8,22 +8,23 @@ 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;
@@ -40,7 +41,7 @@ import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.ActionResponse;
import com.inteligr8.solr.model.BaseResponse;
public abstract class AbstractNodeActionService implements DisposableBean {
public abstract class AbstractNodeActionService {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -53,17 +54,14 @@ public abstract class AbstractNodeActionService implements DisposableBean {
@Autowired
private ExecutorManager executorManager;
@Autowired
private SolrShardHashService shardHashService;
@Autowired(required = false)
@Qualifier(Constants.QUALIFIER_ASIE)
private ShardRegistry shardRegistry;
@Value("${inteligr8.asie.default.concurrentQueueSize}")
@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.default.concurrency}")
@Value("${inteligr8.asie.default.concurrency:16}")
private int concurrency;
protected int getConcurrency() {
@@ -78,22 +76,6 @@ public abstract class AbstractNodeActionService implements DisposableBean {
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
@@ -148,15 +130,16 @@ public abstract class AbstractNodeActionService implements DisposableBean {
}
}
private Future<Void> _action(
final long nodeDbId,
final ActionCallback callback,
Long fullQueueExpireTimeMillis) throws TimeoutException, InterruptedException {
private Future<Void> _action(long nodeDbId, 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.getExecutor();
ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(
this.getThreadNamePrefix(),
this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),
1L, TimeUnit.MINUTES);
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);
@@ -164,12 +147,49 @@ public abstract class AbstractNodeActionService implements DisposableBean {
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 {
actionToShard(nodeDbId, callback, instance);
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()));
}
}
}
} catch (Exception e) {
logger.error("An unexpected exception occurred", e);
logger.error("An exception occurred", e);
callback.error(instance, e.getMessage());
}
return null;
}
};
@@ -184,91 +204,54 @@ public abstract class AbstractNodeActionService implements DisposableBean {
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);
List<com.inteligr8.alfresco.asie.model.ShardInstance> instances = new LinkedList<>();
// 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, Map<Shard, Set<ShardState>>> floc : this.shardRegistry.getFlocs().entrySet()) {
if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE))
continue;
int shardHash = -1;
ShardSet shardset = null;
for (Entry<Shard, Set<ShardState>> 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);
shardHash = this.shardHashService.computeShardInstanceId(shardset, nodeDbId);
}
if (shardHash >= 0) {
if (shard.getKey().getInstance() == shardHash)
instances.add(this.toModel(shardState.getShardInstance(), shardState));
} else {
List<ShardInstance> 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, Map<Shard, Set<ShardState>>> floc : this.shardRegistry.getFlocs().entrySet()) {
if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE))
continue;
for (Entry<Shard, Set<ShardState>> shard : floc.getValue().entrySet()) {
for (ShardState shardState : shard.getValue())
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<Shard, Set<ShardState>> shardsStates = this.shardRegistry.getFlocs().get(floc);
if (shardsStates == null)
throw new IllegalStateException();
Set<ShardState> 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();
@@ -1,10 +1,11 @@
package com.inteligr8.alfresco.asie.service;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
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;
@@ -25,19 +26,19 @@ 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;
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;
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
@Component
public class AcsReconcileService implements DisposableBean {
public class AcsReconcileService implements InitializingBean, DisposableBean {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Logger reconcileLogger = LoggerFactory.getLogger("inteligr8.asie.reconcile");
@@ -54,47 +55,32 @@ public class AcsReconcileService implements DisposableBean {
@Autowired
private SearchService searchService;
@Autowired
private IndexService indexService;
@Autowired
private ReindexService reindexService;
@Autowired
private ExecutorManager executorManager;
@Value("${inteligr8.asie.reconciliation.nodesChunkSize}")
@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
private int nodesChunkSize;
@Value("${inteligr8.asie.reconciliation.nodeTimeoutSeconds}")
@Value("${inteligr8.asie.reconciliation.nodeTimeoutSeconds:10}")
private int nodeTimeoutSeconds;
@Value("${inteligr8.asie.reconciliation.concurrentQueueSize}")
@Value("${inteligr8.asie.reconciliation.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.reconciliation.concurrency}")
@Value("${inteligr8.asie.reconciliation.concurrency:2}")
private int concurrency;
@Value("${inteligr8.asie.reconciliation.waitAfterSolrNodeActionMillis}")
private long waitAfterSolrNodeActionMillis;
@Value("${inteligr8.asie.reconciliation.waitAfterSolrNodeReconcileMillis}")
private long waitAfterSolrNodeReconcileMillis;
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() {
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);
this.executor.shutdown();
}
/**
@@ -106,33 +92,39 @@ 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 reconcile executions. The execution timeouts are
* open up space for new re-index 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.
*
* @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 reindexUnreconciled For nodes not 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.
* @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 reconcile execution should take when dequeued.
* @param execTimeout A timeout for the elapsed time the reindex 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 reconciliation was interrupted (server shutdown).
* @throws InterruptedException The re-index was interrupted (server shutdown).
*/
public void reconcile(
long fromDbId, long toDbId, Integer nodesChunkSize,
boolean indexUnreconciled,
boolean reindexReconciled,
boolean reindexUnreconciled,
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));
this.logger.trace("reconcile({}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
CompositeFuture<Void> future = new CompositeFuture<>();
Future<Void> future = this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit);
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.purge(true);
}
try {
future.get(execTimeout, execUnit);
@@ -141,64 +133,37 @@ 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<Void> reconcile(
long fromDbId, long toDbId, Integer nodesChunkSize,
boolean indexUnreconciled,
boolean reindexReconciled,
boolean reindexUnreconciled,
ReconcileCallback callback) throws InterruptedException {
this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);
if (nodesChunkSize == null)
nodesChunkSize = this.nodesChunkSize;
this.logger.trace("reconcile({}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled);
CompositeFuture<Void> future = new CompositeFuture<>();
try {
return this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null);
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.purge(true);
}
} catch (TimeoutException te) {
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
}
}
protected Future<Void> _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<Void> 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;
}
protected Future<Void> reconcileChunk(
long fromDbId, long toDbId,
boolean indexUnreconciled,
boolean reindexReconciled,
boolean reindexUnreconciled,
ReconcileCallback callback,
long queueTimeout, TimeUnit queueUnit,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
if (this.logger.isTraceEnabled())
this.logger.trace("reconcileChunk({}, {}, {}, {}, {}, {})", fromDbId, toDbId, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
this.logger.trace("reconcileChunk({}, {}, {}, {}, {})", fromDbId, toDbId, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
int dbIdCount = (int) (toDbId - fromDbId);
@@ -232,176 +197,109 @@ public class AcsReconcileService implements DisposableBean {
}
CompositeFuture<Void> future = new CompositeFuture<>();
ThrottledThreadPoolExecutor executor = this.getExecutor();
for (long _nodeDbId = fromDbId; _nodeDbId < toDbId; _nodeDbId++) {
final long nodeDbId = _nodeDbId;
this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId);
Callable<Void> callable;
boolean callingSolr = false;
final int dbIdIndex = (int) (nodeDbId - fromDbId);
if (nodeRefs[dbIdIndex] != null) {
callable = new Callable<Void>() {
@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);
// purposefully forgetting about the returned future
// its results will be logged
// the reconcile thread will continue independently
// the callback will lag
return null;
}
};
if (reindexReconciled)
callingSolr = true;
} else {
callable = new Callable<Void>() {
@Override
public Void call() throws InterruptedException, TimeoutException {
reconcile(nodeDbId, indexUnreconciled, callback);
return null;
}
};
callingSolr = true;
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<Void> callable = new Callable<Void>() {
@Override
public Void call() throws InterruptedException, TimeoutException {
reconcile(nodeDbId, reindexUnreconciled, callback, execTimeout, execUnit);
return null;
}
};
if (queueTimeout < 0L) {
future.combine(executor.submit(callable, -1L, null));
future.combine(this.executor.submit(callable, -1L, null));
} 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);
future.combine(this.executor.submit(callable, queueTimeout, queueUnit));
}
}
return future;
}
public boolean reconcile(long nodeDbId,
boolean index,
ReconcileCallback callback) throws InterruptedException, TimeoutException {
public void reconcile(long nodeDbId,
boolean reindexUnreconciled,
ReconcileCallback callback,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
if (nodeRef == null) {
this.logger.trace("No such ACS node: {}; skipping ...", nodeDbId);
return false;
return;
}
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 false;
return;
}
Set<QName> 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 false;
return;
}
if (!index) {
if (!reindexUnreconciled) {
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);
// purposefully forgetting about the returned future
// its results will be logged
// the reconcile thread will continue independently
// the callback will lag
return true;
logger.debug("A node in the DB is not indexed in Solr; attempt to reindex: {}: {}", nodeDbId, nodeRef);
this.reindex(nodeDbId, nodeRef, callback, execTimeout, execUnit);
}
}
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);
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);
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);
if (callback != null)
callback.processed(nodeDbId,
Collections.emptySet(), Collections.emptySet(),
Collections.singletonMap(instance, message));
}
};
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<>();
Future<Void> future = this.indexService.index(nodeDbId, indexCallback);
if (this.waitAfterSolrNodeActionMillis > 0L) {
Thread.sleep(this.waitAfterSolrNodeActionMillis);
}
return future;
}
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);
if (callback != null)
callback.processed(nodeDbId,
Collections.singleton(instance), Collections.emptySet(),
Collections.emptyMap());
reconcileLogger.info("REINDEXED: {} <=> {}", nodeDbId, nodeRef);
syncHosts.add(instance);
}
@Override
public void scheduled(ShardInstance instance) {
reconcileLogger.info("REINDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance);
if (callback != null)
callback.processed(nodeDbId,
Collections.emptySet(), Collections.singleton(instance),
Collections.emptyMap());
reconcileLogger.info("REINDEXING: {} <=> {}", nodeDbId, nodeRef);
asyncHosts.add(instance);
}
@Override
public void error(ShardInstance instance, String message) {
reconcileLogger.info("FAILED REINDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance);
if (callback != null)
callback.processed(nodeDbId,
Collections.emptySet(), Collections.emptySet(),
Collections.singletonMap(instance, message));
reconcileLogger.info("UNINDEXED: {} <=> {}", nodeDbId, nodeRef);
errorHosts.put(instance, message);
}
};
Future<Void> future = this.reindexService.reindex(nodeDbId, reindexCallback);
if (this.waitAfterSolrNodeActionMillis > 0L) {
Thread.sleep(this.waitAfterSolrNodeActionMillis);
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);
}
return future;
if (callback != null)
callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);
}
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}")
@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
private int nodesChunkSize;
@Override
@@ -6,8 +6,6 @@ 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;
@@ -32,36 +30,42 @@ import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
*/
@Component
public class ExecutorManager implements InitializingBean, DisposableBean, RemovalListener<String, ExecutorService> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Value("${inteligr8.asie.executors.expireTimeInMinutes}")
@Value("${inteligr8.asie.executors.expireTimeInMinutes:30}")
private int expireTimeInMinutes;
private Cache<String, ThrottledThreadPoolExecutor> cache;
private Cache<String, ExecutorService> refCache;
private Cache<String, ExecutorService> expiringCache;
@Override
public void afterPropertiesSet() throws Exception {
this.cache = CacheBuilder.newBuilder()
// 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()
.initialCapacity(8)
.expireAfterAccess(this.expireTimeInMinutes, TimeUnit.MINUTES)
.removalListener(this)
.build();
}
@Override
public void destroy() throws Exception {
this.cache.invalidateAll();
this.cache.cleanUp();
this.refCache.invalidateAll();
this.refCache.cleanUp();
this.expiringCache.invalidateAll();
this.expiringCache.cleanUp();
}
@Override
public void onRemoval(RemovalNotification<String, ExecutorService> notification) {
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());
}
notification.getValue().shutdown();
}
public ThrottledThreadPoolExecutor createThrottled(
@@ -84,10 +88,9 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
final RejectedExecutionHandler rejectedExecutionHandler) {
try {
// if it is already cached, reuse the cache; otherwise create one
return this.cache.get(name, new Callable<ThrottledThreadPoolExecutor>() {
final ExecutorService executor = this.refCache.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,
@@ -100,9 +103,14 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
rejectedExecutionHandler);
}
logger.debug("Created throttled thread pool: {}; threads: {}; queue: {}", name, maximumThreadPoolSize, maximumQueueSize);
executor.prestartAllCoreThreads();
logger.trace("Started {} core threads in thread pool: {}", coreThreadPoolSize, name);
return executor;
}
});
return (ThrottledThreadPoolExecutor) this.expiringCache.get(name, new Callable<ExecutorService>() {
@Override
public ExecutorService call() throws Exception {
return executor;
}
});
@@ -112,7 +120,19 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova
}
public ExecutorService get(String name) {
return this.cache.getIfPresent(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;
}
}
@@ -1,43 +0,0 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.IndexRequest;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
import com.inteligr8.solr.model.Action;
@Component
public class IndexService extends AbstractNodeActionService {
@Override
protected String getActionName() {
return "index";
}
@Override
protected String getThreadNamePrefix() {
return "solr-index";
}
@Override
protected ActionCoreResponse<Action> execute(CoreAdminApi api, String core, long nodeDbId) {
IndexRequest apiRequest = new IndexRequest().withCore(core).withNodeId(nodeDbId);
return api.index(apiRequest);
}
public Future<Void> index(long nodeDbId, ActionCallback callback) throws InterruptedException {
return super.action(nodeDbId, callback);
}
public void index(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
super.action(nodeDbId, callback, fullQueueTimeout, fullQueueUnit, execTimeout, execUnit);
}
}
@@ -1,36 +1,14 @@
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;
import org.alfresco.service.cmr.repository.NodeRef;
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;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.model.ShardSet;
@Component
public class SolrShardHashService {
private final Charset charset = Charset.forName("utf-8");
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private NamespaceService namespaceService;
@Autowired
private NodeService nodeService;
public int hash(Object obj, int shardCount) {
String str = obj.toString();
@@ -45,67 +23,5 @@ public class SolrShardHashService {
hash.add(bytes, 0, bytes.length);
return Math.abs(hash.end()) % shardCount;
}
/**
* @param shardset A ShardSet.
* @param nodeDbId The numeric database ID of an ACS node.
* @return A shard instance from 0 to 1 less than the number of shards; -1 if unable to hash.
*/
public int computeShardInstanceId(ShardSet shardset, long nodeDbId) {
NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
if (nodeRef == null)
throw new AlfrescoRuntimeException("The node " + nodeDbId + " does not exist");
return this.computeShardInstanceId(shardset, nodeRef);
}
/**
* @param shardset A ShardSet.
* @param nodeRef A reference to an ACS node.
* @return A shard instance from 0 to 1 less than the number of shards; -1 if unable to hash.
*/
public int computeShardInstanceId(ShardSet shardset, NodeRef nodeRef) {
switch (shardset.getMethod()) {
case PROPERTY:
QName hashableProperty = QName.createQName(shardset.getPrefixedProperty(), this.namespaceService);
Object fullPropertyValue = this.nodeService.getProperty(nodeRef, hashableProperty);
if (fullPropertyValue == null) {
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;
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()) {
this.logger.debug("Unable to determine shard instance ID because regex pattern doesn't match the hashing property value: {}: {}: {}", nodeRef, shardset.getRegex(), hashableValue);
return -1;
}
hashableValue = matcher.group(1);
this.logger.trace("Extracted shardable value from node: {}: {} => {}", nodeRef, fullPropertyValue, hashableValue);
}
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;
}
}
}
@@ -1,12 +0,0 @@
package com.inteligr8.alfresco.asie.spi;
import com.inteligr8.alfresco.asie.model.ShardInstance;
public interface IndexCallback extends ActionCallback {
@Override
default void unknownResult(ShardInstance instance) {
throw new IllegalStateException();
}
}
@@ -125,9 +125,7 @@ public interface ShardDiscoveryService {
public class ShardedNodeShardStateComparator implements Comparator<Pair<SolrHost, ShardInstanceState>> {
@Override
public int compare(Pair<SolrHost, ShardInstanceState> p1, Pair<SolrHost, ShardInstanceState> p2) {
if (p1 == null) return 1;
else if (p2 == null) return -1;
else return - Long.compare(p1.getSecond().getLastIndexedTxId(), p2.getSecond().getLastIndexedTxId());
return - Long.compare(p1.getSecond().getLastIndexedTxId(), p2.getSecond().getLastIndexedTxId());
}
}
@@ -79,14 +79,12 @@ 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.trace("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 exectuion of the future to complete", unit.toMillis(timeout));
results.add(((RunnableFuture<T>) future).get(timeout, unit));
this.logger.trace("Exectuion completed", unit.toMillis(timeout));
} else {
long remainingTimeMillis = expireTimeMillis - System.currentTimeMillis();
this.logger.trace("Waiting {} ms for the future to complete", remainingTimeMillis);
this.logger.debug("Waiting {} ms for the future to complete", remainingTimeMillis);
results.add(future.get(remainingTimeMillis, TimeUnit.MILLISECONDS));
this.logger.trace("Exectuion completed", unit.toMillis(timeout));
}
}
@@ -126,39 +124,24 @@ 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()) {
Future<T> future = i.next();
if (future.isCancelled()) {
if (includeCancelled) {
this.logger.trace("Removing cancelled future");
removedCancelled++;
i.remove();
} else {
remain++;
}
} else if (future.isDone()) {
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();
}
removedDone++;
i.remove();
} else if (future instanceof CompositeFuture<?>) {
cfutures.add((CompositeFuture<?>) future);
} else {
remain++;
}
}
this.logger.debug("Purged {} cancelled and {} completed futures; {} remain", removedCancelled, removedDone, remain);
this.logger.debug("Purged {} cancelled and {} completed futures", removedCancelled, removedDone);
for (CompositeFuture<?> cfuture : cfutures)
cfuture.purge(includeCancelled);
@@ -80,11 +80,14 @@ 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("Timeout waiting for queue space for runnable");
throw new TimeoutException();
}
return runnable;
@@ -1,63 +0,0 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Index ACS Node in ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Index the specified ACS node in the ASIE indexes.
This call will attempt to index the ACS node on all applicable Solr nodes.
The index operation could be synchronous or asynchronous and could fail on any Solr node.
If any Solr node failed synchronously in the execution, then expect a status code of 500.
The response body will still be identical to the 200/202 status codes.
If any Solr node is executing the index asynchronously and there are no synchronous failures, then expect a status code of 202.</p>
<p>The following path parameters are supported:</p>
<dl>
<dt>nodeId</dt>
<dd>An ACS node ID.</dd>
</dl>
<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>
<pre>
{
"nodeDbId": number,
"success": [
"solrHostSync:8983/solr",
...
],
"scheduled": [
"solrHostAsync:8983/solr",
...
],
"error": [
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
]
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>200</dt>
<dd>OK</dd>
<dt>202</dt>
<dd>Accepted</dd>
<dt>400</dt>
<dd>The path or query parameters are invalid</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/node/{nodeId}/index</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>
@@ -16,13 +16,6 @@
<dt>toDbId</dt>
<dd>A DB ID integer for the ending point of a range, exclusive.</dd>
</dl>
<p>The following query parameters are also supported:</p>
<dl>
<dt>index</dt>
<dd>Schedule an `INDEX` action against all unreconciled nodes.</dd>
<dt>reindex</dt>
<dd>Schedule a `REINDEX` action against all reconciled nodes.</dd>
</dl>
<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>
<pre>
{
@@ -65,7 +58,7 @@
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?index={index?}&amp;reindex={reindex?}</url>
<url>/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?reindex={reindex?}</url>
<format default="json">any</format>
<!-- Security -->
@@ -1,27 +1,8 @@
# 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
inteligr8.asie.reconciliation.waitAfterSolrNodeActionMillis=0
inteligr8.asie.reconciliation.waitAfterSolrNodeReconcileMillis=0
# Action (like indexing and re-indexing) configuration
inteligr8.asie.default.concurrentQueueSize=32
inteligr8.asie.default.concurrency=2
+5 -5
View File
@@ -6,31 +6,31 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.3-SNAPSHOT</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
<groupId>com.inteligr8</groupId>
<artifactId>solr-api</artifactId>
<version>1.2-SNAPSHOT-solr6</version>
<version>1.1.0-solr6</version>
<packaging>jar</packaging>
<name>Apache Solr Jakarta RS API</name>
<properties>
<jackson.version>2.22.1</jackson.version>
<jackson.version>2.18.0</jackson.version>
</properties>
<dependencies>
<dependency>
<groupId>jakarta.annotation</groupId>
<artifactId>jakarta.annotation-api</artifactId>
<version>3.0.0</version>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>jakarta.ws.rs</groupId>
<artifactId>jakarta.ws.rs-api</artifactId>
<version>4.0.0</version>
<version>3.1.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.module</groupId>