MAILBOX-266 ElasticSearch client should be instanciated once

Note : this commit includes a complete rework of how James MPT SMPT tests works.
 - Rely on the guice project (which highlight the initialization problem and that it brakes SMPT)
 - Better manage resources

This was needed because ElasticSearch was never truly configured before. Which you can do with a client provider (you end up never instanciating the client)
but not if you rely on a real client.

mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/DeleteByQueryPerformer.java

@@ -40,18 +40,18 @@ public class DeleteByQueryPerformer {
     public static final int DEFAULT_BATCH_SIZE = 100;
     public static final TimeValue TIMEOUT = new TimeValue(60000);
 
-    private final ClientProvider clientProvider;
+    private final Client client;
     private final ExecutorService executor;
     private final int batchSize;
 
     @Inject
-    public DeleteByQueryPerformer(ClientProvider clientProvider, @Named("AsyncExecutor") ExecutorService executor) {
-        this(clientProvider, executor, DEFAULT_BATCH_SIZE);
+    public DeleteByQueryPerformer(Client client, @Named("AsyncExecutor") ExecutorService executor) {
+        this(client, executor, DEFAULT_BATCH_SIZE);
     }
 
     @VisibleForTesting
-    DeleteByQueryPerformer(ClientProvider clientProvider, @Named("AsyncExecutor") ExecutorService executor, int batchSize) {
-        this.clientProvider = clientProvider;
+    DeleteByQueryPerformer(Client client, @Named("AsyncExecutor") ExecutorService executor, int batchSize) {
+        this.client = client;
         this.executor = executor;
         this.batchSize = batchSize;
     }
@@ -62,17 +62,15 @@ public Void perform(QueryBuilder queryBuilder) {
     }
 
     protected void doDeleteByQuery(QueryBuilder queryBuilder) {
-        try (Client client = clientProvider.get()) {
-            new ScrollIterable(client,
-                client.prepareSearch(ElasticSearchIndexer.MAILBOX_INDEX)
-                    .setTypes(ElasticSearchIndexer.MESSAGE_TYPE)
-                    .setScroll(TIMEOUT)
-                    .setNoFields()
-                    .setQuery(queryBuilder)
-                    .setSize(batchSize))
-                .stream()
-                .forEach(searchResponse -> deleteRetrievedIds(client, searchResponse));
-        }
+        new ScrollIterable(client,
+            client.prepareSearch(ElasticSearchIndexer.MAILBOX_INDEX)
+                .setTypes(ElasticSearchIndexer.MESSAGE_TYPE)
+                .setScroll(TIMEOUT)
+                .setNoFields()
+                .setQuery(queryBuilder)
+                .setSize(batchSize))
+            .stream()
+            .forEach(searchResponse -> deleteRetrievedIds(client, searchResponse));
     }
 
     private ListenableActionFuture<BulkResponse> deleteRetrievedIds(Client client, SearchResponse searchResponse) {

mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIndexer.java

@@ -56,40 +56,34 @@ public String getUpdatedDocumentPart() {
     public static final String MAILBOX_INDEX = "mailbox";
     public static final String MESSAGE_TYPE = "message";
 
-    private final ClientProvider clientProvider;
+    private final Client client;
     private final DeleteByQueryPerformer deleteByQueryPerformer;
 
     @Inject
-    public ElasticSearchIndexer(ClientProvider clientProvider, DeleteByQueryPerformer deleteByQueryPerformer) {
-        this.clientProvider = clientProvider;
+    public ElasticSearchIndexer(Client client, DeleteByQueryPerformer deleteByQueryPerformer) {
+        this.client = client;
         this.deleteByQueryPerformer = deleteByQueryPerformer;
     }
 
     public IndexResponse indexMessage(String id, String content) {
         checkArgument(content);
-        try (Client client = clientProvider.get()) {
-            return client.prepareIndex(MAILBOX_INDEX, MESSAGE_TYPE, id)
-                .setSource(content)
-                .get();
-        }
+        return client.prepareIndex(MAILBOX_INDEX, MESSAGE_TYPE, id)
+            .setSource(content)
+            .get();
     }
 
     public BulkResponse updateMessages(List<UpdatedRepresentation> updatedDocumentParts) {
         Preconditions.checkNotNull(updatedDocumentParts);
-        try (Client client = clientProvider.get()) {
-            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
-            updatedDocumentParts.forEach(updatedDocumentPart -> bulkRequestBuilder.add(client.prepareUpdate(MAILBOX_INDEX, MESSAGE_TYPE, updatedDocumentPart.getId())
-                .setDoc(updatedDocumentPart.getUpdatedDocumentPart())));
-            return bulkRequestBuilder.get();
-        }
+        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+        updatedDocumentParts.forEach(updatedDocumentPart -> bulkRequestBuilder.add(client.prepareUpdate(MAILBOX_INDEX, MESSAGE_TYPE, updatedDocumentPart.getId())
+            .setDoc(updatedDocumentPart.getUpdatedDocumentPart())));
+        return bulkRequestBuilder.get();
     }
-    
+
     public BulkResponse deleteMessages(List<String> ids) {
-        try (Client client = clientProvider.get()) {
-            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
-            ids.forEach(id -> bulkRequestBuilder.add(client.prepareDelete(MAILBOX_INDEX, MESSAGE_TYPE, id)));
-            return bulkRequestBuilder.get();
-        }
+        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+        ids.forEach(id -> bulkRequestBuilder.add(client.prepareDelete(MAILBOX_INDEX, MESSAGE_TYPE, id)));
+        return bulkRequestBuilder.get();
     }
 
     public Void deleteAllMatchingQuery(QueryBuilder queryBuilder) {

mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/IndexCreationFactory.java

@@ -36,33 +36,31 @@ public class IndexCreationFactory {
     private static final int DEFAULT_NB_REPLICA = 0;
     public static final String CASE_INSENSITIVE = "case_insensitive";
 
-    public static ClientProvider createIndex(ClientProvider clientProvider, int nbShards, int nbReplica) {
+    public static Client createIndex(Client client, int nbShards, int nbReplica) {
         try {
-            return createIndex(clientProvider, generateSetting(nbShards, nbReplica));
+            return createIndex(client, generateSetting(nbShards, nbReplica));
         } catch (IOException e) {
             LOGGER.error("Error while creating index : ", e);
-            return clientProvider;
+            return client;
         }
     }
 
-    public static ClientProvider createIndex(ClientProvider clientProvider) {
-        return createIndex(clientProvider, DEFAULT_NB_SHARDS, DEFAULT_NB_REPLICA);
+    public static Client createIndex(Client client) {
+        return createIndex(client, DEFAULT_NB_SHARDS, DEFAULT_NB_REPLICA);
     }
 
-    private static ClientProvider createIndex(ClientProvider clientProvider, XContentBuilder settings) {
+    private static Client createIndex(Client client, XContentBuilder settings) {
         try {
-            try (Client client = clientProvider.get()) {
                 client.admin()
                     .indices()
                     .prepareCreate(ElasticSearchIndexer.MAILBOX_INDEX)
                     .setSettings(settings)
                     .execute()
                     .actionGet();
-            }
         } catch (IndexAlreadyExistsException exception) {
             LOGGER.info("Index [" + ElasticSearchIndexer.MAILBOX_INDEX + "] already exist");
         }
-        return clientProvider;
+        return client;
     }
 
     private static XContentBuilder generateSetting(int nbShards, int nbReplica) throws IOException {

mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/NodeMappingFactory.java

@@ -45,21 +45,19 @@ public class NodeMappingFactory {
     public static final String RAW = "raw";
     public static final String ANALYZER = "analyzer";
 
-    public static ClientProvider applyMapping(ClientProvider clientProvider) {
-        return applyMapping(clientProvider, getMappingContent());
+    public static Client applyMapping(Client client) {
+        return applyMapping(client, getMappingContent());
     }
 
-    public static ClientProvider applyMapping(ClientProvider clientProvider, XContentBuilder mappingsSources) {
-        try (Client client = clientProvider.get()) {
-            client.admin()
-                .indices()
-                .preparePutMapping(ElasticSearchIndexer.MAILBOX_INDEX)
-                .setType(ElasticSearchIndexer.MESSAGE_TYPE)
-                .setSource(mappingsSources)
-                .execute()
-                .actionGet();
-        }
-        return clientProvider;
+    public static Client applyMapping(Client client, XContentBuilder mappingsSources) {
+        client.admin()
+            .indices()
+            .preparePutMapping(ElasticSearchIndexer.MAILBOX_INDEX)
+            .setType(ElasticSearchIndexer.MESSAGE_TYPE)
+            .setSource(mappingsSources)
+            .execute()
+            .actionGet();
+        return client;
     }
 
     private static XContentBuilder getMappingContent() {

mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java

@@ -26,7 +26,6 @@
 
 import javax.inject.Inject;
 
-import org.apache.james.mailbox.elasticsearch.ClientProvider;
 import org.apache.james.mailbox.elasticsearch.ElasticSearchIndexer;
 import org.apache.james.mailbox.elasticsearch.json.JsonMessageConstants;
 import org.apache.james.mailbox.elasticsearch.query.QueryConverter;
@@ -49,27 +48,25 @@ public class ElasticSearchSearcher<Id extends MailboxId> {
     private static final TimeValue TIMEOUT = new TimeValue(60000);
     public static final int DEFAULT_SIZE = 100;
 
-    private final ClientProvider clientProvider;
+    private final Client client;
     private final QueryConverter queryConverter;
     private final int size;
 
     @Inject
-    public ElasticSearchSearcher(ClientProvider clientProvider, QueryConverter queryConverter) {
-        this(clientProvider, queryConverter, DEFAULT_SIZE);
+    public ElasticSearchSearcher(Client client, QueryConverter queryConverter) {
+        this(client, queryConverter, DEFAULT_SIZE);
     }
 
-    public ElasticSearchSearcher(ClientProvider clientProvider, QueryConverter queryConverter, int size) {
-        this.clientProvider = clientProvider;
+    public ElasticSearchSearcher(Client client, QueryConverter queryConverter, int size) {
+        this.client = client;
         this.queryConverter = queryConverter;
         this.size = size;
     }
 
     public Iterator<Long> search(Mailbox<Id> mailbox, SearchQuery searchQuery) throws MailboxException {
-        try (Client client = clientProvider.get()) {
-            return new ScrollIterable(client, getSearchRequestBuilder(client, mailbox, searchQuery)).stream()
-                .flatMap(this::transformResponseToUidStream)
-                .iterator();
-        }
+        return new ScrollIterable(client, getSearchRequestBuilder(client, mailbox, searchQuery)).stream()
+            .flatMap(this::transformResponseToUidStream)
+            .iterator();
     }
 
     private SearchRequestBuilder getSearchRequestBuilder(Client client, Mailbox<Id> mailbox, SearchQuery searchQuery) {

mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIndexerTest.java

@@ -56,14 +56,14 @@ public class ElasticSearchIndexerTest {
     public void setup() throws IOException {
         node = embeddedElasticSearch.getNode();
         TestingClientProvider clientProvider = new TestingClientProvider(node);
-        DeleteByQueryPerformer deleteByQueryPerformer = new DeleteByQueryPerformer(clientProvider, Executors.newSingleThreadExecutor(), MINIMUM_BATCH_SIZE) {
+        DeleteByQueryPerformer deleteByQueryPerformer = new DeleteByQueryPerformer(clientProvider.get(), Executors.newSingleThreadExecutor(), MINIMUM_BATCH_SIZE) {
             @Override
             public Void perform(QueryBuilder queryBuilder) {
                 doDeleteByQuery(queryBuilder);
                 return null;
             }
         };
-        testee = new ElasticSearchIndexer(clientProvider, deleteByQueryPerformer);
+        testee = new ElasticSearchIndexer(clientProvider.get(), deleteByQueryPerformer);
     }
 
     @Test

mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java

@@ -49,6 +49,7 @@
 import org.apache.james.mailbox.store.StoreMailboxManager;
 import org.apache.james.mailbox.store.StoreMessageManager;
 import org.apache.james.mailbox.store.mail.model.Mailbox;
+import org.elasticsearch.client.Client;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -165,13 +166,13 @@ public void setUp() throws Exception {
     }
 
     private void initializeMailboxManager() throws Exception {
-        ClientProvider clientProvider = NodeMappingFactory.applyMapping(
-            IndexCreationFactory.createIndex(new TestingClientProvider(embeddedElasticSearch.getNode()))
+        Client client = NodeMappingFactory.applyMapping(
+            IndexCreationFactory.createIndex(new TestingClientProvider(embeddedElasticSearch.getNode()).get())
         );
         MailboxSessionMapperFactory<InMemoryId> mapperFactory = new InMemoryMailboxSessionMapperFactory();
         elasticSearchListeningMessageSearchIndex = new ElasticSearchListeningMessageSearchIndex<>(mapperFactory,
-            new ElasticSearchIndexer(clientProvider, new DeleteByQueryPerformer(clientProvider, Executors.newSingleThreadExecutor(), BATCH_SIZE)),
-            new ElasticSearchSearcher<>(clientProvider, new QueryConverter(new CriterionConverter()), SEARCH_SIZE),
+            new ElasticSearchIndexer(client, new DeleteByQueryPerformer(client, Executors.newSingleThreadExecutor(), BATCH_SIZE)),
+            new ElasticSearchSearcher<>(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE),
             new MessageToElasticSearchJson(new DefaultTextExtractor(), ZoneId.of("Europe/Paris")));
         storeMailboxManager = new InMemoryMailboxManager(
             mapperFactory,

mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/search/ScrollIterableTest.java

@@ -62,9 +62,9 @@ public class ScrollIterableTest {
     @Before
     public void setUp() throws Exception {
         clientProvider = new TestingClientProvider(embeddedElasticSearch.getNode());
-        IndexCreationFactory.createIndex(clientProvider);
+        IndexCreationFactory.createIndex(clientProvider.get());
         embeddedElasticSearch.awaitForElasticSearch();
-        NodeMappingFactory.applyMapping(clientProvider, getMappingsSources());
+        NodeMappingFactory.applyMapping(clientProvider.get(), getMappingsSources());
     }
 
     private XContentBuilder getMappingsSources() throws IOException {

mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java

@@ -32,7 +32,6 @@
 import org.apache.james.mailbox.acl.MailboxACLResolver;
 import org.apache.james.mailbox.acl.SimpleGroupMembershipResolver;
 import org.apache.james.mailbox.acl.UnionMailboxACLResolver;
-import org.apache.james.mailbox.elasticsearch.ClientProvider;
 import org.apache.james.mailbox.elasticsearch.DeleteByQueryPerformer;
 import org.apache.james.mailbox.elasticsearch.ElasticSearchIndexer;
 import org.apache.james.mailbox.elasticsearch.EmbeddedElasticSearch;
@@ -58,6 +57,7 @@
 import org.apache.james.mpt.api.ImapFeatures.Feature;
 import org.apache.james.mpt.host.JamesImapHostSystem;
 import org.apache.james.mpt.imapmailbox.MailboxCreationDelegate;
+import org.elasticsearch.client.Client;
 
 import com.google.common.base.Throwables;
 
@@ -95,17 +95,17 @@ protected void resetData() throws Exception {
     }
 
     private void initFields() {
-        ClientProvider clientProvider = NodeMappingFactory.applyMapping(
-            IndexCreationFactory.createIndex(new TestingClientProvider(embeddedElasticSearch.getNode()))
+        Client client = NodeMappingFactory.applyMapping(
+            IndexCreationFactory.createIndex(new TestingClientProvider(embeddedElasticSearch.getNode()).get())
         );
 
         userManager = new MockAuthenticator();
         InMemoryMailboxSessionMapperFactory factory = new InMemoryMailboxSessionMapperFactory();
 
         ElasticSearchListeningMessageSearchIndex<InMemoryId> searchIndex = new ElasticSearchListeningMessageSearchIndex<>(
             factory,
-            new ElasticSearchIndexer(clientProvider, new DeleteByQueryPerformer(clientProvider, Executors.newSingleThreadExecutor())),
-            new ElasticSearchSearcher<>(clientProvider, new QueryConverter(new CriterionConverter())),
+            new ElasticSearchIndexer(client, new DeleteByQueryPerformer(client, Executors.newSingleThreadExecutor())),
+            new ElasticSearchSearcher<>(client, new QueryConverter(new CriterionConverter())),
             new MessageToElasticSearchJson(new DefaultTextExtractor()));
 
         MailboxACLResolver aclResolver = new UnionMailboxACLResolver();