CR: add test for bulk rejection behaviour

original-brownbear · original-brownbear · commit a3c8cac75e13 · 2019-03-24T21:12:11.000+01:00
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java b/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
@@ -171,6 +171,7 @@ protected void doRun() {
 
             @Override
             public void onRejection(Exception e) {
+                // Fail all operations after a bulk rejection hit an action that waited for a mapping update and finish the request
                 while (context.hasMoreOperationsToExecute()) {
                     context.setRequestToExecute(context.getCurrent());
                     final long version = context.getRequestToExecute().version();
@@ -179,7 +180,6 @@ public void onRejection(Exception e) {
                         : primary.getFailedIndexResult(e, version);
                     onComplete(result, context, null);
                 }
-                // We're done, there's no more operations to execute so we resolve the wrapped listener
                 finishRequest();
             }
 
diff --git a/server/src/test/java/org/elasticsearch/action/bulk/BulkRejectionIT.java b/server/src/test/java/org/elasticsearch/action/bulk/BulkRejectionIT.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.action.bulk;
+
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.Collections;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 2)
+public class BulkRejectionIT extends ESIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put("thread_pool.write.size", 1)
+            .put("thread_pool.write.queue_size", 1)
+            .build();
+    }
+
+    @Override
+    protected int numberOfReplicas() {
+        return 1;
+    }
+
+    protected int numberOfShards() {
+        return 5;
+    }
+
+    public void testBulkRejectionAfterDynamicMappingUpdate() throws Exception {
+        final String index = "test";
+        assertAcked(prepareCreate(index));
+        ensureGreen();
+        final BulkRequest request1 = new BulkRequest();
+        for (int i = 0; i < 500; ++i) {
+            request1.add(new IndexRequest(index).source(Collections.singletonMap("key" + i, "value" + i)))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        }
+        // Huge request to keep the write pool busy so that requests waiting on a mapping update in the other bulk request get rejected
+        // by the write pool
+        final BulkRequest request2 = new BulkRequest();
+        for (int i = 0; i < 10_000; ++i) {
+            request2.add(new IndexRequest(index).source(Collections.singletonMap("key", "valuea" + i)))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        }
+        final Client client = client();
+        final ActionFuture<BulkResponse> bulkFuture1 = client.bulk(request1);
+        final ActionFuture<BulkResponse> bulkFuture2 = client.bulk(request2);
+        bulkFuture1.actionGet();
+        bulkFuture2.actionGet();
+        internalCluster().assertSeqNos();
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -171,6 +171,7 @@ protected void doRun() {`
`171`	`171`
`172`	`172`	`@Override`
`173`	`173`	`public void onRejection(Exception e) {`
	`174`	`+ // Fail all operations after a bulk rejection hit an action that waited for a mapping update and finish the request`
`174`	`175`	`while (context.hasMoreOperationsToExecute()) {`
`175`	`176`	`context.setRequestToExecute(context.getCurrent());`
`176`	`177`	`final long version = context.getRequestToExecute().version();`
`@@ -179,7 +180,6 @@ public void onRejection(Exception e) {`
`179`	`180`	`: primary.getFailedIndexResult(e, version);`
`180`	`181`	`onComplete(result, context, null);`
`181`	`182`	`}`
`182`		`- // We're done, there's no more operations to execute so we resolve the wrapped listener`
`183`	`183`	`finishRequest();`
`184`	`184`	`}`
`185`	`185`