Decouple BulkProcessor from client implementation

client/rest-high-level/src/test/java/org/elasticsearch/client/CrudIT.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
@@ -38,6 +39,9 @@
 import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.VersionType;
@@ -46,10 +50,13 @@
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.Collections.singletonMap;
 
@@ -577,6 +584,102 @@ public void testBulk() throws IOException {
         assertTrue(bulkResponse.getTookInMillis() > 0);
         assertEquals(nbItems, bulkResponse.getItems().length);
 
+        validateBulkResponses(nbItems, errors, bulkResponse, bulkRequest);
+    }
+
+    public void testBulkProcessorIntegration() throws IOException, InterruptedException {
+        int nbItems = randomIntBetween(10, 100);
+        boolean[] errors = new boolean[nbItems];
+
+        XContentType xContentType = randomFrom(XContentType.JSON, XContentType.SMILE);
+
+        AtomicReference<BulkResponse> responseRef = new AtomicReference<>();
+        AtomicReference<BulkRequest> requestRef = new AtomicReference<>();
+        AtomicReference<Throwable> error = new AtomicReference<>();
+
+        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
+            @Override
+            public void beforeBulk(long executionId, BulkRequest request) {
+
+            }
+
+            @Override
+            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
+                responseRef.set(response);
+                requestRef.set(request);
+            }
+
+            @Override
+            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
+                error.set(failure);
+            }
+        };
+
+        ThreadPool threadPool = new ThreadPool(Settings.builder().put("node.name", getClass().getName()).build());
+        try(BulkProcessor processor = new BulkProcessor.Builder(highLevelClient()::bulkAsync, listener, threadPool)
+            .setConcurrentRequests(0)
+            .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.GB))
+            .setBulkActions(nbItems + 1)
+            .build()) {
+            for (int i = 0; i < nbItems; i++) {
+                String id = String.valueOf(i);
+                boolean erroneous = randomBoolean();
+                errors[i] = erroneous;
+
+                DocWriteRequest.OpType opType = randomFrom(DocWriteRequest.OpType.values());
+                if (opType == DocWriteRequest.OpType.DELETE) {
+                    if (erroneous == false) {
+                        assertEquals(RestStatus.CREATED,
+                            highLevelClient().index(new IndexRequest("index", "test", id).source("field", -1)).status());
+                    }
+                    DeleteRequest deleteRequest = new DeleteRequest("index", "test", id);
+                    processor.add(deleteRequest);
+
+                } else {
+                    if (opType == DocWriteRequest.OpType.INDEX) {
+                        IndexRequest indexRequest = new IndexRequest("index", "test", id).source(xContentType, "id", i);
+                        if (erroneous) {
+                            indexRequest.version(12L);
+                        }
+                        processor.add(indexRequest);
+
+                    } else if (opType == DocWriteRequest.OpType.CREATE) {
+                        IndexRequest createRequest = new IndexRequest("index", "test", id).source(xContentType, "id", i).create(true);
+                        if (erroneous) {
+                            assertEquals(RestStatus.CREATED, highLevelClient().index(createRequest).status());
+                        }
+                        processor.add(createRequest);
+
+                    } else if (opType == DocWriteRequest.OpType.UPDATE) {
+                        UpdateRequest updateRequest = new UpdateRequest("index", "test", id)
+                            .doc(new IndexRequest().source(xContentType, "id", i));
+                        if (erroneous == false) {
+                            assertEquals(RestStatus.CREATED,
+                                highLevelClient().index(new IndexRequest("index", "test", id).source("field", -1)).status());
+                        }
+                        processor.add(updateRequest);
+                    }
+                }
+            }
+            assertNull(responseRef.get());
+            assertNull(requestRef.get());
+        }
+
+
+        BulkResponse bulkResponse = responseRef.get();
+        BulkRequest bulkRequest = requestRef.get();
+
+        assertEquals(RestStatus.OK, bulkResponse.status());
+        assertTrue(bulkResponse.getTookInMillis() > 0);
+        assertEquals(nbItems, bulkResponse.getItems().length);
+        assertNull(error.get());
+
+        validateBulkResponses(nbItems, errors, bulkResponse, bulkRequest);
+
+        terminate(threadPool);
+    }
+
+    private void validateBulkResponses(int nbItems, boolean[] errors, BulkResponse bulkResponse, BulkRequest bulkRequest) {
         for (int i = 0; i < nbItems; i++) {
             BulkItemResponse bulkItemResponse = bulkResponse.getItems()[i];
 

core/src/main/java/org/elasticsearch/action/bulk/BulkProcessor.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
@@ -28,17 +29,14 @@
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
-import org.elasticsearch.common.util.concurrent.FutureUtils;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.Closeable;
 import java.util.Objects;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
 
 /**
  * A bulk processor is a thread safe bulk processing class, allowing to easily set when to "flush" a new bulk request
@@ -66,7 +64,7 @@ public interface Listener {
 
         /**
          * Callback after a failed execution of bulk request.
-         *
+         * <p>
          * Note that in case an instance of <code>InterruptedException</code> is passed, which means that request processing has been
          * cancelled externally, the thread's interruption status has been restored prior to calling this method.
          */
@@ -78,10 +76,10 @@ public interface Listener {
      */
     public static class Builder {
 
-        private final Client client;
+        private final BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer;
         private final Listener listener;
+        private final ThreadPool threadPool;
 
-        private String name;
         private int concurrentRequests = 1;
         private int bulkActions = 1000;
         private ByteSizeValue bulkSize = new ByteSizeValue(5, ByteSizeUnit.MB);
@@ -92,17 +90,10 @@ public static class Builder {
          * Creates a builder of bulk processor with the client to use and the listener that will be used
          * to be notified on the completion of bulk requests.
          */
-        public Builder(Client client, Listener listener) {
-            this.client = client;
+        public Builder(BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer, Listener listener, ThreadPool threadPool) {
+            this.consumer = consumer;
             this.listener = listener;
-        }
-
-        /**
-         * Sets an optional name to identify this bulk processor.
-         */
-        public Builder setName(String name) {
-            this.name = name;
-            return this;
+            this.threadPool = threadPool;
         }
 
         /**
@@ -164,23 +155,21 @@ public Builder setBackoffPolicy(BackoffPolicy backoffPolicy) {
          * Builds a new bulk processor.
          */
         public BulkProcessor build() {
-            return new BulkProcessor(client, backoffPolicy, listener, name, concurrentRequests, bulkActions, bulkSize, flushInterval);
+            return new BulkProcessor(consumer, backoffPolicy, listener, concurrentRequests, bulkActions, bulkSize, flushInterval, threadPool);
         }
     }
 
     public static Builder builder(Client client, Listener listener) {
         Objects.requireNonNull(client, "client");
         Objects.requireNonNull(listener, "listener");
 
-        return new Builder(client, listener);
+        return new Builder(client::bulk, listener, client.threadPool());
     }
 
     private final int bulkActions;
     private final long bulkSize;
 
-
-    private final ScheduledThreadPoolExecutor scheduler;
-    private final ScheduledFuture<?> scheduledFuture;
+    private final ThreadPool.Cancellable cancellableFlushTask;
 
     private final AtomicLong executionIdGen = new AtomicLong();
 
@@ -189,22 +178,21 @@ public static Builder builder(Client client, Listener listener) {
 
     private volatile boolean closed = false;
 
-    BulkProcessor(Client client, BackoffPolicy backoffPolicy, Listener listener, @Nullable String name, int concurrentRequests, int bulkActions, ByteSizeValue bulkSize, @Nullable TimeValue flushInterval) {
+    BulkProcessor(BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer, BackoffPolicy backoffPolicy, Listener listener,
+                  int concurrentRequests, int bulkActions, ByteSizeValue bulkSize, @Nullable TimeValue flushInterval,
+                  ThreadPool threadPool) {
         this.bulkActions = bulkActions;
         this.bulkSize = bulkSize.getBytes();
-
         this.bulkRequest = new BulkRequest();
-        this.bulkRequestHandler = (concurrentRequests == 0) ? BulkRequestHandler.syncHandler(client, backoffPolicy, listener) : BulkRequestHandler.asyncHandler(client, backoffPolicy, listener, concurrentRequests);
 
-        if (flushInterval != null) {
-            this.scheduler = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1, EsExecutors.daemonThreadFactory(client.settings(), (name != null ? "[" + name + "]" : "") + "bulk_processor"));
-            this.scheduler.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
-            this.scheduler.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
-            this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(new Flush(), flushInterval.millis(), flushInterval.millis(), TimeUnit.MILLISECONDS);
+        if (concurrentRequests == 0) {
+            this.bulkRequestHandler = BulkRequestHandler.syncHandler(consumer, backoffPolicy, listener, threadPool);
         } else {
-            this.scheduler = null;
-            this.scheduledFuture = null;
+            this.bulkRequestHandler = BulkRequestHandler.asyncHandler(consumer, backoffPolicy, listener, threadPool, concurrentRequests);
         }
+
+        // Start period flushing task after everything is setup
+        this.cancellableFlushTask = startFlushTask(flushInterval, threadPool);
     }
 
     /**
@@ -214,20 +202,20 @@ public static Builder builder(Client client, Listener listener) {
     public void close() {
         try {
             awaitClose(0, TimeUnit.NANOSECONDS);
-        } catch(InterruptedException exc) {
+        } catch (InterruptedException exc) {
             Thread.currentThread().interrupt();
         }
     }
 
     /**
      * Closes the processor. If flushing by time is enabled, then it's shutdown. Any remaining bulk actions are flushed.
-     *
+     * <p>
      * If concurrent requests are not enabled, returns {@code true} immediately.
      * If concurrent requests are enabled, waits for up to the specified timeout for all bulk requests to complete then returns {@code true},
      * If the specified waiting time elapses before all bulk requests complete, {@code false} is returned.
      *
      * @param timeout The maximum time to wait for the bulk requests to complete
-     * @param unit The time unit of the {@code timeout} argument
+     * @param unit    The time unit of the {@code timeout} argument
      * @return {@code true} if all bulk requests completed and {@code false} if the waiting time elapsed before all the bulk requests completed
      * @throws InterruptedException If the current thread is interrupted
      */
@@ -236,10 +224,9 @@ public synchronized boolean awaitClose(long timeout, TimeUnit unit) throws Inter
             return true;
         }
         closed = true;
-        if (this.scheduledFuture != null) {
-            FutureUtils.cancel(this.scheduledFuture);
-            this.scheduler.shutdown();
-        }
+
+        this.cancellableFlushTask.cancel();
+
         if (bulkRequest.numberOfActions() > 0) {
             execute();
         }
@@ -301,12 +288,28 @@ public BulkProcessor add(BytesReference data, @Nullable String defaultIndex, @Nu
      * Adds the data from the bytes to be processed by the bulk processor
      */
     public synchronized BulkProcessor add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType,
-                                  @Nullable String defaultPipeline, @Nullable Object payload, XContentType xContentType) throws Exception {
+                                          @Nullable String defaultPipeline, @Nullable Object payload, XContentType xContentType) throws Exception {
         bulkRequest.add(data, defaultIndex, defaultType, null, null, null, defaultPipeline, payload, true, xContentType);
         executeIfNeeded();
         return this;
     }
 
+    private ThreadPool.Cancellable startFlushTask(TimeValue flushInterval, ThreadPool threadPool) {
+        if (flushInterval == null) {
+            return new ThreadPool.Cancellable() {
+                @Override
+                public void cancel() {}
+
+                @Override
+                public boolean isCancelled() {
+                    return true;
+                }
+            };
+        }
+
+        return threadPool.scheduleWithFixedDelay(new Flush(), flushInterval, ThreadPool.Names.GENERIC);
+    }
+
     private void executeIfNeeded() {
         ensureOpen();
         if (!isOverTheLimit()) {