Removing BulkProcessor

Author: masseyke

qa/multi-cluster-search/src/test/java/org/elasticsearch/search/CCSDuelIT.java

@@ -17,7 +17,7 @@
 import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
-import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkProcessor2;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
@@ -32,6 +32,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.index.query.InnerHitBuilder;
@@ -194,9 +195,9 @@ private static void indexDocuments(String idPrefix) throws IOException, Interrup
         response = createIndex(INDEX_NAME, settings, mapping);
         assertTrue(response.isAcknowledged());
 
-        BulkProcessor bulkProcessor = BulkProcessor.builder(
+        BulkProcessor2 bulkProcessor = BulkProcessor2.builder(
             (r, l) -> restHighLevelClient.bulkAsync(r, RequestOptions.DEFAULT, l),
-            new BulkProcessor.Listener() {
+            new BulkProcessor2.Listener() {
                 @Override
                 public void beforeBulk(long executionId, BulkRequest request) {}
 
@@ -206,11 +207,11 @@ public void afterBulk(long executionId, BulkRequest request, BulkResponse respon
                 }
 
                 @Override
-                public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
+                public void afterBulk(long executionId, BulkRequest request, Exception failure) {
                     throw new AssertionError("Failed to execute bulk", failure);
                 }
             },
-            "CCSDuelIT"
+            new DeterministicTaskQueue(random()).getThreadPool()
         ).build();
 
         int numQuestions = randomIntBetween(50, 100);