global parameters on bulk processor

pgomulka · pgomulka · commit 61a93181c667 · 2018-10-25T18:40:07.000+02:00
bulk api documentation changes
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/BulkProcessorIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/BulkProcessorIT.java
@@ -20,7 +20,6 @@
 package org.elasticsearch.client;
 
 import com.carrotsearch.randomizedtesting.generators.RandomPicks;
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkProcessor;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -32,7 +31,6 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
@@ -283,6 +281,41 @@ public void testBulkProcessorConcurrentRequestsReadOnlyIndex() throws Exception
         assertMultiGetResponse(highLevelClient().mget(multiGetRequest, RequestOptions.DEFAULT), testDocs);
     }
 
+    @SuppressWarnings("unchecked")
+    public void testGlobalParametersAndSingleRequest() throws Exception {
+        createIndexWithMultipleShards("test");
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        BulkProcessorTestListener listener = new BulkProcessorTestListener(latch);
+        createFieldAddingPipleine("pipeline_id", "fieldNameXYZ", "valueXYZ");
+
+        // tag::bulk-processor-mix-parameters
+        try (BulkProcessor processor = initBulkProcessorBuilder(listener)
+                .setGlobalIndex("tweets")
+                .setGlobalType("_doc")
+                .setGlobalRouting("routing")
+                .setGlobalPipeline("pipeline_id")
+                .build()) {
+
+
+            processor.add(new IndexRequest() // <1>
+                .source(XContentType.JSON, "user", "some user"));
+            processor.add(new IndexRequest("blogs", "post_type", "1") // <2>
+                .source(XContentType.JSON, "title", "some title"));
+        }
+        // end::bulk-request-mix-pipeline
+        latch.await();
+
+        Iterable<SearchHit> hits = searchAll(new SearchRequest("tweets").routing("routing"));
+        assertThat(hits, everyItem(hasProperty(fieldFromSource("user"), equalTo("some user"))));
+        assertThat(hits, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
+
+
+        Iterable<SearchHit> blogs = searchAll(new SearchRequest("blogs").routing("routing"));
+        assertThat(blogs, everyItem(hasProperty(fieldFromSource("title"), equalTo("some title"))));
+        assertThat(blogs, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
+    }
+
     @SuppressWarnings("unchecked")
     public void testGlobalParametersAndBulkProcessor() throws Exception {
         createIndexWithMultipleShards("test");
@@ -424,13 +457,5 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
         }
     }
 
-    private void createIndexWithMultipleShards(String index) throws IOException {
-        CreateIndexRequest indexRequest = new CreateIndexRequest(index);
-        int shards = randomIntBetween(2,10);
-        indexRequest.settings(Settings.builder()
-            .put("index.number_of_shards", shards)
-            .put("index.number_of_replicas", 0)
-        );
-        highLevelClient().indices().create(indexRequest, RequestOptions.DEFAULT);
-    }
+
 }
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/BulkRequestWithGlobalParametersIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/BulkRequestWithGlobalParametersIT.java
@@ -19,12 +19,10 @@
 
 package org.elasticsearch.client;
 
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.search.SearchHit;
 
@@ -89,16 +87,17 @@ public void testMixPipelineOnRequestAndGlobal() throws IOException {
         createFieldAddingPipleine("globalId", "fieldXYZ", "valueXYZ");
         createFieldAddingPipleine("perIndexId", "someNewField", "someValue");
 
+        // tag::bulk-request-mix-pipeline
         BulkRequest request = new BulkRequest();
         request.pipeline("globalId");
 
         request.add(new IndexRequest("test", "doc", "1")
             .source(XContentType.JSON, "field", "bulk1")
-            .setPipeline("perIndexId"));
+            .setPipeline("perIndexId")); // <1>
 
         request.add(new IndexRequest("test", "doc", "2")
-            .source(XContentType.JSON, "field", "bulk2"));
-
+            .source(XContentType.JSON, "field", "bulk2")); // <2>
+        // end::bulk-request-mix-pipeline
         bulk(request);
 
         Iterable<SearchHit> hits = searchAll("test");
@@ -215,15 +214,4 @@ private BulkResponse bulk(BulkRequest request) throws IOException {
     private static <T> Function<SearchHit, T> fieldFromSource(String fieldName) {
         return (response) -> (T) response.getSourceAsMap().get(fieldName);
     }
-
-    private void createIndexWithMultipleShards(String index) throws IOException {
-        CreateIndexRequest indexRequest = new CreateIndexRequest(index);
-        // keeping shard # high to make sure different routings will lead to different shards - reduces the chances of random failure
-        int shards = randomIntBetween(8,10);
-        indexRequest.settings(Settings.builder()
-            .put("index.number_of_shards", shards)
-            .put("index.number_of_replicas", 0)
-        );
-        highLevelClient().indices().create(indexRequest, RequestOptions.DEFAULT);
-    }
 }
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/ESRestHighLevelClientTestCase.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/ESRestHighLevelClientTestCase.java
@@ -21,6 +21,7 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
@@ -195,4 +196,14 @@ protected void refreshIndexes(String... indices) throws IOException {
         Response refreshResponse = client().performRequest(new Request("POST", "/" + joinedIndices + "/_refresh"));
         assertEquals(200, refreshResponse.getStatusLine().getStatusCode());
     }
+
+    protected void createIndexWithMultipleShards(String index) throws IOException {
+        CreateIndexRequest indexRequest = new CreateIndexRequest(index);
+        int shards = randomIntBetween(8,10);
+        indexRequest.settings(Settings.builder()
+            .put("index.number_of_shards", shards)
+            .put("index.number_of_replicas", 0)
+        );
+        highLevelClient().indices().create(indexRequest, RequestOptions.DEFAULT);
+    }
 }
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/CRUDDocumentationIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/CRUDDocumentationIT.java
@@ -748,6 +748,16 @@ public void testBulk() throws Exception {
             request.waitForActiveShards(2); // <1>
             request.waitForActiveShards(ActiveShardCount.ALL); // <2>
             // end::bulk-request-active-shards
+            // tag::bulk-request-pipeline
+            request.pipeline("pipelineId"); // <1>
+            // end::bulk-request-pipeline
+            // tag::bulk-request-routing
+            request.routing("routingId"); // <1>
+            // end::bulk-request-routing
+
+            // tag::bulk-request-index-type
+            BulkRequest requestWithDefaultedIndexAndType = new BulkRequest("posts","_doc"); // <1>
+            // end::bulk-request-index-type
 
             // tag::bulk-execute-listener
             ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
diff --git a/docs/java-api/docs/bulk.asciidoc b/docs/java-api/docs/bulk.asciidoc
@@ -165,3 +165,27 @@ client.admin().indices().prepareRefresh().get();
 client.prepareSearch().get();
 --------------------------------------------------
 
+
+[[java-docs-bulk-global-parameters]]
+==== Global Parameters
+
+Similarly to REST API where you can provide URL parameters, you can specify global parameters on BulkRequest as well
+ as BulkProcessor. These global parameters serve as defaults and can be overriden by local parameters specified on
+ each sub request. Some parameters have to be set before any sub request is added - index, type - and you have to specify them
+ during BulkRequest or BulkProcessor creation. Some are optional - pipeline, routing - and can be specified at any point before the bulk is
+ sent.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/BulkProcessorIT.java[bulk-processor-mix-parameters]
+--------------------------------------------------
+<1> global parameters from the BulkRequest will be applied on a sub request
+<2> local pipeline parameter on a sub request will override global parameters from BulkRequest
+
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/BulkRequestWithGlobalParametersIT.java[bulk-request-mix-pipeline]
+--------------------------------------------------
+<1> local pipeline parameter on a sub request will override global pipeline from the BulkRequest
+<2> global parameter from the BulkRequest will be applied on a sub request
diff --git a/docs/java-rest/high-level/document/bulk.asciidoc b/docs/java-rest/high-level/document/bulk.asciidoc
@@ -62,6 +62,24 @@ the index/update/delete operations.
 <2> Number of shard copies provided as a `ActiveShardCount`: can be `ActiveShardCount.ALL`,
 `ActiveShardCount.ONE` or `ActiveShardCount.DEFAULT` (default)
 
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/CRUDDocumentationIT.java[bulk-request-pipeline]
+--------------------------------------------------
+<1> Global pipelineId used on all sub requests, unless overridden on a sub request
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/CRUDDocumentationIT.java[bulk-request-routing]
+--------------------------------------------------
+<1> Global routingId used on all sub requests, unless overridden on a sub request
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/CRUDDocumentationIT.java[bulk-request-index-type]
+--------------------------------------------------
+<1> A bulk request with global index and type used on all sub requests, unless overridden on a sub request.
+Both parameters are @Nullable and can only be set during BulkRequest creation.
 
 [[java-rest-high-document-bulk-sync]]
 ==== Synchronous Execution
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
@@ -342,13 +342,13 @@ public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Null
                 String index = defaultIndex;
                 String type = defaultType;
                 String id = null;
-                String routing = defaultRouting;
+                String routing = valueOrDefault(defaultRouting, globalRouting);
                 FetchSourceContext fetchSourceContext = defaultFetchSourceContext;
                 String opType = null;
                 long version = Versions.MATCH_ANY;
                 VersionType versionType = VersionType.INTERNAL;
                 int retryOnConflict = 0;
-                String pipeline = defaultPipeline;
+                String pipeline = valueOrDefault(defaultPipeline, globalPipeline);
 
                 // at this stage, next token can either be END_OBJECT (and use default index and type, with auto generated id)
                 // or START_OBJECT which will have another set of parameters
@@ -520,8 +520,8 @@ public final BulkRequest pipeline(String globalPipeline) {
         return this;
     }
 
-    public final BulkRequest routing(String defaultRouting){
-        this.globalRouting = defaultRouting;
+    public final BulkRequest routing(String globalRouting){
+        this.globalRouting = globalRouting;
         return this;
     }
     /**
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestBuilder.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestBuilder.java
@@ -41,6 +41,10 @@
 public class BulkRequestBuilder extends ActionRequestBuilder<BulkRequest, BulkResponse>
         implements WriteRequestBuilder<BulkRequestBuilder> {
 
+    public BulkRequestBuilder(ElasticsearchClient client, BulkAction action, @Nullable String globalIndex, @Nullable String globalType) {
+        super(client, action, new BulkRequest(globalIndex, globalType));
+    }
+
     public BulkRequestBuilder(ElasticsearchClient client, BulkAction action) {
         super(client, action, new BulkRequest());
     }
@@ -153,4 +157,14 @@ public final BulkRequestBuilder setTimeout(String timeout) {
     public int numberOfActions() {
         return request.numberOfActions();
     }
+
+    public BulkRequestBuilder pipeline(String globalPipeline) {
+        request.pipeline(globalPipeline);
+        return this;
+    }
+
+    public BulkRequestBuilder routing(String globalRouting) {
+        request.routing(globalRouting);
+        return this;
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/client/Client.java b/server/src/main/java/org/elasticsearch/client/Client.java
@@ -232,6 +232,11 @@ public interface Client extends ElasticsearchClient, Releasable {
      */
     BulkRequestBuilder prepareBulk();
 
+    /**
+     * Executes a bulk of index / delete operations with default index and/or type
+     */
+    BulkRequestBuilder prepareBulk(@Nullable String globalIndex, @Nullable String globalType);
+
     /**
      * Gets the document that was indexed from an index with a type and id.
      *
diff --git a/server/src/main/java/org/elasticsearch/client/support/AbstractClient.java b/server/src/main/java/org/elasticsearch/client/support/AbstractClient.java
@@ -471,6 +471,11 @@ public BulkRequestBuilder prepareBulk() {
         return new BulkRequestBuilder(this, BulkAction.INSTANCE);
     }
 
+    @Override
+    public BulkRequestBuilder prepareBulk(@Nullable String globalIndex, @Nullable String globalType) {
+        return new BulkRequestBuilder(this, BulkAction.INSTANCE, globalIndex, globalType);
+    }
+
     @Override
     public ActionFuture<GetResponse> get(final GetRequest request) {
         return execute(GetAction.INSTANCE, request);
diff --git a/server/src/test/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java b/server/src/test/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java
@@ -20,22 +20,40 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.ingest.IngestTestPlugin;
+import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESIntegTestCase;
 
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.test.StreamsUtils.copyToStringFromClasspath;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 
 public class BulkIntegrationIT extends ESIntegTestCase {
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IngestTestPlugin.class);
+    }
+
     public void testBulkIndexCreatesMapping() throws Exception {
         String bulkAction = copyToStringFromClasspath("/org/elasticsearch/action/bulk/bulk-log.json");
         BulkRequestBuilder bulkBuilder = client().prepareBulk();
@@ -81,4 +99,52 @@ public void testBulkWithWriteIndexAndRouting() {
         assertFalse(bulkResponse.hasFailures());
         assertFalse(client().prepareGet("index3", "type", "id").setRouting("1").get().isExists());
     }
+
+    public void testBulkWithGlobalDefaults() throws Exception {
+        // all requests in the json are missing index and type parameters: "_index" : "test", "_type" : "type1",
+        String bulkAction = copyToStringFromClasspath("/org/elasticsearch/action/bulk/simple-bulk-missing-index-type.json");
+        {
+            BulkRequestBuilder bulkBuilder = client().prepareBulk();
+            bulkBuilder.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, null, XContentType.JSON);
+            ActionRequestValidationException ex = expectThrows(ActionRequestValidationException.class, bulkBuilder::get);
+
+            assertThat(ex.validationErrors(), containsInAnyOrder(
+                "index is missing",
+                "index is missing",
+                "index is missing",
+                "type is missing",
+                "type is missing",
+                "type is missing"));
+        }
+
+        {
+            createSamplePipeline("pipeline");
+            BulkRequestBuilder bulkBuilder = client().prepareBulk("test","type1")
+                .routing("routing")
+                .pipeline("pipeline");
+
+            bulkBuilder.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, null, XContentType.JSON);
+            BulkResponse bulkItemResponses = bulkBuilder.get();
+            assertFalse(bulkItemResponses.hasFailures());
+        }
+    }
+
+    private void createSamplePipeline(String pipelineId) throws IOException, ExecutionException, InterruptedException {
+        XContentBuilder pipeline = jsonBuilder()
+            .startObject()
+                .startArray("processors")
+                    .startObject()
+                        .startObject("test")
+                        .endObject()
+                    .endObject()
+                .endArray()
+            .endObject();
+
+        AcknowledgedResponse acknowledgedResponse = client().admin()
+            .cluster()
+            .putPipeline(new PutPipelineRequest(pipelineId, BytesReference.bytes(pipeline), XContentType.JSON))
+            .get();
+
+        assertTrue(acknowledgedResponse.isAcknowledged());
+    }
 }
diff --git a/server/src/test/resources/org/elasticsearch/action/bulk/simple-bulk-missing-index-type.json b/server/src/test/resources/org/elasticsearch/action/bulk/simple-bulk-missing-index-type.json
@@ -0,0 +1,5 @@
+{ "index":{"_id":"1"} }
+{ "field1" : "value1" }
+{ "delete" : {  "_id" : "2" } }
+{ "create" : {  "_id" : "3" } }
+{ "field1" : "value3" }
