Repository: james-project Updated Branches: refs/heads/master 6951b14b1 -> 990a3bf96
JAMES-1706 retry and wait for elasticsearch and cassandra to be up and running Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/990a3bf9 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/990a3bf9 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/990a3bf9 Branch: refs/heads/master Commit: 990a3bf965c995da7c29372c0d5fb32c722ee013 Parents: 6951b14 Author: Fabien Vignon <[email protected]> Authored: Thu Mar 10 18:59:23 2016 +0100 Committer: Matthieu Baechler <[email protected]> Committed: Thu Mar 17 09:48:10 2016 +0100 ---------------------------------------------------------------------- .../guice/destination/conf/cassandra.properties | 4 +- .../destination/conf/elasticsearch.properties | 2 + server/container/cassandra-guice/pom.xml | 4 ++ .../modules/mailbox/CassandraSessionModule.java | 40 ++++++++++++---- .../mailbox/ElasticSearchMailboxModule.java | 27 +++++++++-- .../ScheduledExecutorServiceProvider.java | 49 ++++++++++++++++++++ server/pom.xml | 5 ++ 7 files changed, 116 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/dockerfiles/run/guice/destination/conf/cassandra.properties ---------------------------------------------------------------------- diff --git a/dockerfiles/run/guice/destination/conf/cassandra.properties b/dockerfiles/run/guice/destination/conf/cassandra.properties index c6a7342..c9a94bb 100644 --- a/dockerfiles/run/guice/destination/conf/cassandra.properties +++ b/dockerfiles/run/guice/destination/conf/cassandra.properties @@ -3,4 +3,6 @@ cassandra.ip=cassandra cassandra.port=9042 cassandra.keyspace=apache_james -cassandra.replication.factor=1 \ No newline at end of file +cassandra.replication.factor=1 +cassandra.retryConnection.maxRetries=10 +cassandra.retryConnection.minDelay=5000 \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/dockerfiles/run/guice/destination/conf/elasticsearch.properties ---------------------------------------------------------------------- diff --git a/dockerfiles/run/guice/destination/conf/elasticsearch.properties b/dockerfiles/run/guice/destination/conf/elasticsearch.properties index 2eec953..7ed13a4 100644 --- a/dockerfiles/run/guice/destination/conf/elasticsearch.properties +++ b/dockerfiles/run/guice/destination/conf/elasticsearch.properties @@ -24,3 +24,5 @@ elasticsearch.masterHost=elasticsearch elasticsearch.port=9300 elasticsearch.nb.shards=1 elasticsearch.nb.replica=0 +elasticsearch.retryConnection.maxRetries=7 +elasticsearch.retryConnection.minDelay=3000 http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/server/container/cassandra-guice/pom.xml ---------------------------------------------------------------------- diff --git a/server/container/cassandra-guice/pom.xml b/server/container/cassandra-guice/pom.xml index 957904c..09b76d4 100644 --- a/server/container/cassandra-guice/pom.xml +++ b/server/container/cassandra-guice/pom.xml @@ -383,6 +383,10 @@ <scope>test</scope> </dependency> <dependency> + <groupId>com.nurkiewicz.asyncretry</groupId> + <artifactId>asyncretry</artifactId> + </dependency> + <dependency> <groupId>commons-daemon</groupId> <artifactId>commons-daemon</artifactId> </dependency> http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraSessionModule.java ---------------------------------------------------------------------- diff --git a/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraSessionModule.java b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraSessionModule.java index c3588ba..f40cf41 100644 --- a/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraSessionModule.java +++ b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraSessionModule.java @@ -20,6 +20,8 @@ package org.apache.james.modules.mailbox; import java.io.FileNotFoundException; import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledExecutorService; import org.apache.commons.configuration.ConfigurationException; import org.apache.commons.configuration.PropertiesConfiguration; @@ -32,14 +34,20 @@ import org.apache.james.filesystem.api.FileSystem; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.Session; +import com.datastax.driver.core.exceptions.NoHostAvailableException; import com.google.inject.AbstractModule; import com.google.inject.Provides; import com.google.inject.Singleton; +import com.nurkiewicz.asyncretry.AsyncRetryExecutor; public class CassandraSessionModule extends AbstractModule { + private static final int DEFAULT_CONNECTION_MAX_RETRIES = 10; + private static final int DEFAULT_CONNECTION_MIN_DELAY = 5000; + @Override protected void configure() { + bind(ScheduledExecutorService.class).toProvider(ScheduledExecutorServiceProvider.class); } @Provides @@ -59,19 +67,33 @@ public class CassandraSessionModule extends AbstractModule { @Provides @Singleton - Cluster provideCluster(FileSystem fileSystem) throws FileNotFoundException, ConfigurationException { + Cluster provideCluster(FileSystem fileSystem, AsyncRetryExecutor executor) throws FileNotFoundException, ConfigurationException, ExecutionException, InterruptedException { PropertiesConfiguration configuration = getConfiguration(fileSystem); - - return ClusterWithKeyspaceCreatedFactory.clusterWithInitializedKeyspace( - ClusterFactory.createClusterForSingleServerWithoutPassWord( - configuration.getString("cassandra.ip"), - configuration.getInt("cassandra.port")), - configuration.getString("cassandra.keyspace"), - configuration.getInt("cassandra.replication.factor")); + + return getRetryer(executor, configuration) + .getWithRetry(ctx -> ClusterWithKeyspaceCreatedFactory.clusterWithInitializedKeyspace( + ClusterFactory.createClusterForSingleServerWithoutPassWord( + configuration.getString("cassandra.ip"), + configuration.getInt("cassandra.port")), + configuration.getString("cassandra.keyspace"), + configuration.getInt("cassandra.replication.factor"))) + .get(); + } + + private static AsyncRetryExecutor getRetryer(AsyncRetryExecutor executor, PropertiesConfiguration configuration) { + return executor.retryOn(NoHostAvailableException.class) + .withProportionalJitter() + .withMaxRetries(configuration.getInt("cassandra.retryConnection.maxRetries", DEFAULT_CONNECTION_MAX_RETRIES)) + .withMinDelay(configuration.getInt("cassandra.retryConnection.minDelay", DEFAULT_CONNECTION_MIN_DELAY)); + } + + @Provides + private AsyncRetryExecutor provideAsyncRetryExecutor(ScheduledExecutorService scheduler) { + return new AsyncRetryExecutor(scheduler); } private PropertiesConfiguration getConfiguration(FileSystem fileSystem) throws FileNotFoundException, ConfigurationException { return new PropertiesConfiguration(fileSystem.getFile(FileSystem.FILE_PROTOCOL_AND_CONF + "cassandra.properties")); } - + } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java ---------------------------------------------------------------------- diff --git a/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java index c9a3301..7fc7638 100644 --- a/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java +++ b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java @@ -20,6 +20,7 @@ package org.apache.james.modules.mailbox; import java.io.FileNotFoundException; +import java.util.concurrent.ExecutionException; import javax.inject.Singleton; @@ -35,13 +36,18 @@ import org.apache.james.mailbox.elasticsearch.events.ElasticSearchListeningMessa import org.apache.james.mailbox.store.extractor.TextExtractor; import org.apache.james.mailbox.store.search.MessageSearchIndex; import org.apache.james.mailbox.tika.extractor.TikaTextExtractor; +import org.elasticsearch.client.transport.NoNodeAvailableException; import com.google.inject.AbstractModule; import com.google.inject.Provides; import com.google.inject.TypeLiteral; +import com.nurkiewicz.asyncretry.AsyncRetryExecutor; public class ElasticSearchMailboxModule extends AbstractModule { + private static final int DEFAULT_CONNECTION_MAX_RETRIES = 7; + private static final int DEFAULT_CONNECTION_MIN_DELAY = 3000; + @Override protected void configure() { bind(new TypeLiteral<MessageSearchIndex<CassandraId>>(){}).to(new TypeLiteral<ElasticSearchListeningMessageSearchIndex<CassandraId>>() {}); @@ -50,15 +56,26 @@ public class ElasticSearchMailboxModule extends AbstractModule { @Provides @Singleton - protected ClientProvider provideClientProvider(FileSystem fileSystem) throws ConfigurationException, FileNotFoundException { + protected ClientProvider provideClientProvider(FileSystem fileSystem, AsyncRetryExecutor executor) throws ConfigurationException, FileNotFoundException, ExecutionException, InterruptedException { PropertiesConfiguration propertiesReader = new PropertiesConfiguration(fileSystem.getFile(FileSystem.FILE_PROTOCOL_AND_CONF + "elasticsearch.properties")); + ClientProvider clientProvider = new ClientProviderImpl(propertiesReader.getString("elasticsearch.masterHost"), - propertiesReader.getInt("elasticsearch.port")); - IndexCreationFactory.createIndex(clientProvider, - propertiesReader.getInt("elasticsearch.nb.shards"), - propertiesReader.getInt("elasticsearch.nb.replica")); + propertiesReader.getInt("elasticsearch.port")); + getRetryer(executor, propertiesReader) + .getWithRetry(ctx -> IndexCreationFactory.createIndex(clientProvider, + propertiesReader.getInt("elasticsearch.nb.shards"), + propertiesReader.getInt("elasticsearch.nb.replica"))) + .get(); NodeMappingFactory.applyMapping(clientProvider); return clientProvider; } + private static AsyncRetryExecutor getRetryer(AsyncRetryExecutor executor, PropertiesConfiguration configuration) { + return executor + .withProportionalJitter() + .retryOn(NoNodeAvailableException.class) + .withMaxRetries(configuration.getInt("elasticsearch.retryConnection.maxRetries", DEFAULT_CONNECTION_MAX_RETRIES)) + .withMinDelay(configuration.getInt("elasticsearch.retryConnection.minDelay", DEFAULT_CONNECTION_MIN_DELAY)); + } + } http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ScheduledExecutorServiceProvider.java ---------------------------------------------------------------------- diff --git a/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ScheduledExecutorServiceProvider.java b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ScheduledExecutorServiceProvider.java new file mode 100644 index 0000000..48fe3f1 --- /dev/null +++ b/server/container/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ScheduledExecutorServiceProvider.java @@ -0,0 +1,49 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ + +package org.apache.james.modules.mailbox; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +import javax.annotation.PreDestroy; + +import com.google.common.annotations.VisibleForTesting; +import com.google.inject.Provider; + +@VisibleForTesting +class ScheduledExecutorServiceProvider implements Provider<ScheduledExecutorService> { + + private final ScheduledExecutorService scheduler; + + @VisibleForTesting + ScheduledExecutorServiceProvider() { + scheduler = Executors.newSingleThreadScheduledExecutor(); + } + + @Override + public ScheduledExecutorService get() { + return scheduler; + } + + @PreDestroy + private void stop() { + scheduler.shutdown(); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/990a3bf9/server/pom.xml ---------------------------------------------------------------------- diff --git a/server/pom.xml b/server/pom.xml index a10084d..358c385 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -1304,6 +1304,11 @@ <version>4.0</version> </dependency> <dependency> + <groupId>com.nurkiewicz.asyncretry</groupId> + <artifactId>asyncretry</artifactId> + <version>0.0.7</version> + </dependency> + <dependency> <groupId>org.apache.onami.lifecycle</groupId> <artifactId>org.apache.onami.lifecycle.jsr250</artifactId> <version>0.2.0-SNAPSHOT</version> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
