Reject empty IDs

When indexing a document via the bulk API where IDs can be explicitly
specified, we currently accept an empty ID. This is problematic because
such a document can not be obtained via the get API. Instead, we should
rejected these requets as accepting them could be a dangerous form of
leniency. Additionally, we already have a way of specifying
auto-generated IDs and that is to not explicitly specify an ID so we do
not need a second way. This commit rejects the individual requests where
ID is specified but empty.

Relates #24118

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

@@ -308,7 +308,7 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
 
             assertEquals(RestStatus.BAD_REQUEST, exception.status());
             assertEquals("Elasticsearch exception [type=illegal_argument_exception, " +
-                         "reason=Can't specify parent if no parent field has been configured]", exception.getMessage());
+                         "reason=can't specify parent if no parent field has been configured]", exception.getMessage());
         }
         {
             ElasticsearchStatusException exception = expectThrows(ElasticsearchStatusException.class, () -> {

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

@@ -279,7 +279,7 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
                             break;
                         default: throw new AssertionError("request type not supported: [" + docWriteRequest.opType() + "]");
                     }
-                } catch (ElasticsearchParseException | RoutingMissingException e) {
+                } catch (ElasticsearchParseException | IllegalArgumentException | RoutingMissingException e) {
                     BulkItemResponse.Failure failure = new BulkItemResponse.Failure(concreteIndex.getName(), docWriteRequest.type(), docWriteRequest.id(), e);
                     BulkItemResponse bulkItemResponse = new BulkItemResponse(i, docWriteRequest.opType(), failure);
                     responses.set(i, bulkItemResponse);

core/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -491,14 +491,18 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
             }
 
             if (parent != null && !mappingMd.hasParentField()) {
-                throw new IllegalArgumentException("Can't specify parent if no parent field has been configured");
+                throw new IllegalArgumentException("can't specify parent if no parent field has been configured");
             }
         } else {
             if (parent != null) {
-                throw new IllegalArgumentException("Can't specify parent if no parent field has been configured");
+                throw new IllegalArgumentException("can't specify parent if no parent field has been configured");
             }
         }
 
+        if ("".equals(id)) {
+            throw new IllegalArgumentException("if _id is specified it must not be empty");
+        }
+
         // generate id if not already provided
         if (id == null) {
             assert autoGeneratedTimestamp == -1 : "timestamp has already been generated!";

core/src/test/java/org/elasticsearch/search/child/ChildQuerySearchIT.java

@@ -1182,13 +1182,13 @@ public class ChildQuerySearchIT extends ESIntegTestCase {
             client().prepareIndex("test", "child1", "c1").setParent("p1").setSource("c_field", "blue").get();
             fail();
         } catch (IllegalArgumentException e) {
-            assertThat(e.toString(), containsString("Can't specify parent if no parent field has been configured"));
+            assertThat(e.toString(), containsString("can't specify parent if no parent field has been configured"));
         }
         try {
             client().prepareIndex("test", "child2", "c2").setParent("p1").setSource("c_field", "blue").get();
             fail();
         } catch (IllegalArgumentException e) {
-            assertThat(e.toString(), containsString("Can't specify parent if no parent field has been configured"));
+            assertThat(e.toString(), containsString("can't specify parent if no parent field has been configured"));
         }
 
         refresh();

modules/reindex/src/test/java/org/elasticsearch/index/reindex/ReindexParentChildTests.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.index.reindex;
 
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
+import org.elasticsearch.action.bulk.byscroll.BulkByScrollResponse;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryBuilder;
 
@@ -27,7 +28,10 @@ import static org.elasticsearch.index.query.QueryBuilders.hasParentQuery;
 import static org.elasticsearch.index.query.QueryBuilders.idsQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchHits;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasToString;
+import static org.hamcrest.Matchers.instanceOf;
 
 /**
  * Index-by-search tests for parent/child.
@@ -76,12 +80,11 @@ public class ReindexParentChildTests extends ReindexTestCase {
         createParentChildDocs("source");
 
         ReindexRequestBuilder copy = reindex().source("source").destination("dest").filter(findsCity);
-        try {
-            copy.get();
-            fail("Expected exception");
-        } catch (IllegalArgumentException e) {
-            assertThat(e.getMessage(), equalTo("Can't specify parent if no parent field has been configured"));
-        }
+        final BulkByScrollResponse response = copy.get();
+        assertThat(response.getBulkFailures().size(), equalTo(1));
+        final Exception cause = response.getBulkFailures().get(0).getCause();
+        assertThat(cause, instanceOf(IllegalArgumentException.class));
+        assertThat(cause, hasToString(containsString("can't specify parent if no parent field has been configured")));
     }
 
     /**

rest-api-spec/src/main/resources/rest-api-spec/test/bulk/10_basic.yaml

@@ -23,3 +23,38 @@
 
   - match: {count: 2}
 
+---
+"Empty _id":
+  - skip:
+      version: " - 5.3.0"
+      reason: empty IDs were not rejected until 5.3.1
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: test
+              _type: type
+              _id: ''
+          - f: 1
+          - index:
+              _index: test
+              _type: type
+              _id: id
+          - f: 2
+          - index:
+              _index: test
+              _type: type
+          - f: 3
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: illegal_argument_exception }
+  - match: { items.0.index.error.reason: if _id is specified it must not be empty }
+  - match: { items.1.index.created: true }
+  - match: { items.2.index.created: true }
+
+  - do:
+      count:
+        index: test
+
+  - match: { count: 2 }