Compare commits
21
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ff331cac48 | ||
|
|
d86c96fcb1 | ||
|
|
933a786038 | ||
|
|
bb183c5e50 | ||
|
|
f909f3429b | ||
|
|
fe2eaa0588 | ||
|
|
68471be9ef | ||
|
|
82410805db | ||
|
|
ceb8d2c690 | ||
|
|
1c15853652 | ||
|
|
cf1bc8d2be | ||
|
|
a49ddbd373 | ||
|
|
738e5b98f8 | ||
|
|
9d9d17c49b | ||
|
|
43109e7d9c | ||
|
|
dcb37c3dc8 | ||
|
|
68ffb56cc9 | ||
|
|
dc436e9ce0 | ||
|
|
18c39b0fc5 | ||
|
|
b7aa615103 | ||
|
|
e8f133b873 |
+23
-24
@@ -6,17 +6,22 @@
|
||||
<parent>
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-platform-module-parent</artifactId>
|
||||
<version>1.3-SNAPSHOT</version>
|
||||
<version>1.3.1</version>
|
||||
<relativePath>../</relativePath>
|
||||
</parent>
|
||||
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-api</artifactId>
|
||||
<version>1.2-SNAPSHOT-asie2</version>
|
||||
<version>1.1.1-asie2</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>ASIE Jakarta RS API</name>
|
||||
<description>Alfresco Search & Insight Engine Jakarta RS API</description>
|
||||
|
||||
<properties>
|
||||
<alfresco.platform.version>23.2.0</alfresco.platform.version>
|
||||
</properties>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
@@ -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>
|
||||
|
||||
@@ -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.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")
|
||||
|
||||
+1
-1
@@ -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());
|
||||
|
||||
+1
-1
@@ -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() {
|
||||
+1
-1
@@ -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() {
|
||||
@@ -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
@@ -6,7 +6,7 @@
|
||||
<parent>
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-platform-module-parent</artifactId>
|
||||
<version>1.3-SNAPSHOT</version>
|
||||
<version>1.3.1</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
-1
@@ -264,7 +264,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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-platform-module-parent</artifactId>
|
||||
<version>1.3-SNAPSHOT</version>
|
||||
<version>1.3.1</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
@@ -6,7 +6,7 @@
|
||||
<parent>
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-platform-module-parent</artifactId>
|
||||
<version>1.3-SNAPSHOT</version>
|
||||
<version>1.3.1</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.1.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;
|
||||
|
||||
+4
-4
@@ -44,10 +44,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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+10
-2
@@ -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() {
|
||||
|
||||
+82
-99
@@ -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
|
||||
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 (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 {
|
||||
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();
|
||||
|
||||
|
||||
+96
-139
@@ -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,6 +26,7 @@ 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;
|
||||
@@ -37,7 +39,7 @@ 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");
|
||||
@@ -60,41 +62,29 @@ public class AcsReconcileService implements DisposableBean {
|
||||
@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;
|
||||
private ThrottledThreadPoolExecutor executor;
|
||||
|
||||
@Value("${inteligr8.asie.reconciliation.waitAfterSolrNodeReconcileMillis}")
|
||||
private long waitAfterSolrNodeReconcileMillis;
|
||||
@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,21 +96,20 @@ 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,
|
||||
@@ -129,10 +118,18 @@ public class AcsReconcileService implements DisposableBean {
|
||||
ReconcileCallback callback,
|
||||
long queueTimeout, TimeUnit queueUnit,
|
||||
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
|
||||
if (nodesChunkSize == null)
|
||||
nodesChunkSize = this.nodesChunkSize;
|
||||
if (this.logger.isTraceEnabled())
|
||||
this.logger.trace("reconcile({}, {}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
|
||||
|
||||
Future<Void> future = this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit);
|
||||
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);
|
||||
}
|
||||
|
||||
try {
|
||||
future.get(execTimeout, execUnit);
|
||||
@@ -141,50 +138,25 @@ 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,
|
||||
ReconcileCallback callback) throws InterruptedException {
|
||||
this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);
|
||||
|
||||
try {
|
||||
return this._reconcile(fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null);
|
||||
} catch (TimeoutException te) {
|
||||
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
|
||||
}
|
||||
}
|
||||
|
||||
protected Future<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;
|
||||
this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);
|
||||
|
||||
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);
|
||||
try {
|
||||
for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {
|
||||
long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
|
||||
future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null));
|
||||
future.purge(true);
|
||||
}
|
||||
} catch (TimeoutException te) {
|
||||
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
|
||||
}
|
||||
|
||||
return future;
|
||||
@@ -232,14 +204,12 @@ 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) {
|
||||
@@ -251,157 +221,144 @@ public class AcsReconcileService implements DisposableBean {
|
||||
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
|
||||
reindex(nodeDbId, nodeRefs[dbIdIndex], callback, execTimeout, execUnit);
|
||||
return null;
|
||||
}
|
||||
};
|
||||
if (reindexReconciled)
|
||||
callingSolr = true;
|
||||
} else {
|
||||
callable = new Callable<Void>() {
|
||||
@Override
|
||||
public Void call() throws InterruptedException, TimeoutException {
|
||||
reconcile(nodeDbId, indexUnreconciled, callback);
|
||||
reconcile(nodeDbId, indexUnreconciled, callback, execTimeout, execUnit);
|
||||
return null;
|
||||
}
|
||||
};
|
||||
callingSolr = true;
|
||||
}
|
||||
|
||||
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,
|
||||
public void reconcile(long nodeDbId,
|
||||
boolean index,
|
||||
ReconcileCallback callback) throws InterruptedException, TimeoutException {
|
||||
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) {
|
||||
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 index: {}: {}", nodeDbId, nodeRef);
|
||||
this.index(nodeDbId, nodeRef, callback, execTimeout, execUnit);
|
||||
}
|
||||
}
|
||||
|
||||
public Future<Void> index(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException, TimeoutException {
|
||||
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<>();
|
||||
|
||||
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());
|
||||
reconcileLogger.info("INDEXED: {} <=> {}", nodeDbId, nodeRef);
|
||||
syncHosts.add(instance);
|
||||
}
|
||||
|
||||
@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());
|
||||
reconcileLogger.info("INDEXING: {} <=> {}", nodeDbId, nodeRef);
|
||||
asyncHosts.add(instance);
|
||||
}
|
||||
|
||||
@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));
|
||||
reconcileLogger.info("FAILED INDEX: {} <=> {}", nodeDbId, nodeRef);
|
||||
errorHosts.put(instance, message);
|
||||
}
|
||||
};
|
||||
|
||||
Future<Void> future = this.indexService.index(nodeDbId, indexCallback);
|
||||
|
||||
if (this.waitAfterSolrNodeActionMillis > 0L) {
|
||||
Thread.sleep(this.waitAfterSolrNodeActionMillis);
|
||||
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);
|
||||
}
|
||||
|
||||
return future;
|
||||
if (callback != null)
|
||||
callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);
|
||||
}
|
||||
|
||||
public Future<Void> reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException {
|
||||
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<>();
|
||||
|
||||
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("FAILED REINDEX: {} <=> {}", 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;
|
||||
@@ -33,35 +31,41 @@ 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,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();
|
||||
@@ -46,66 +24,4 @@ public class SolrShardHashService {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
+4
-1
@@ -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;
|
||||
|
||||
-19
@@ -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
@@ -6,31 +6,31 @@
|
||||
<parent>
|
||||
<groupId>com.inteligr8.alfresco</groupId>
|
||||
<artifactId>asie-platform-module-parent</artifactId>
|
||||
<version>1.3-SNAPSHOT</version>
|
||||
<version>1.3.1</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>
|
||||
|
||||
Reference in New Issue
Block a user