Reindex: Better error message for pipeline in wrong place (#21985)

`_update_by_query` supports specifying the `pipeline` to process the
documents as a url parameter but `_reindex` doesn't. It doesn't because
everything about the `_reindex` request that has to do with writing
the documents is grouped under the `dest` object in the request body.
This changes the response parameter from
`request [_reindex] contains unrecognized parameter: [pipeline]` to
`_reindex doesn't support [pipeline] as a query parmaeter. Specify it in the [dest] object instead.`

modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestDeleteByQueryAction.java

@@ -48,14 +48,14 @@ public class RestDeleteByQueryAction extends AbstractBulkByQueryRestHandler<Dele
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        if (false == request.hasContent()) {
-            throw new ElasticsearchException("_delete_by_query requires a request body");
-        }
         return doPrepareRequest(request, client, false, false);
     }
 
     @Override
     protected DeleteByQueryRequest buildRequest(RestRequest request) throws IOException {
+        if (false == request.hasContent()) {
+            throw new ElasticsearchException("_delete_by_query requires a request body");
+        }
         /*
          * Passing the search request through DeleteByQueryRequest first allows
          * it to set its own defaults which differ from SearchRequest's

modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.index.reindex;
 
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.node.NodeClient;
@@ -113,14 +112,18 @@ public class RestReindexAction extends AbstractBaseReindexRestHandler<ReindexReq
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        if (false == request.hasContent()) {
-            throw new ElasticsearchException("_reindex requires a request body");
-        }
         return doPrepareRequest(request, client, true, true);
     }
 
     @Override
     protected ReindexRequest buildRequest(RestRequest request) throws IOException {
+        if (false == request.hasContent()) {
+            throw new IllegalArgumentException("_reindex requires a request body");
+        }
+        if (request.hasParam("pipeline")) {
+            throw new IllegalArgumentException("_reindex doesn't support [pipeline] as a query parmaeter. "
+                    + "Specify it in the [dest] object instead.");
+        }
         ReindexRequest internal = new ReindexRequest(new SearchRequest(), new IndexRequest());
         try (XContentParser xcontent = XContentFactory.xContent(request.content()).createParser(request.content())) {
             PARSER.parse(xcontent, internal, new ReindexParseContext(searchRequestParsers, parseFieldMatcher));

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

@@ -23,20 +23,25 @@ import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.common.ParseFieldMatcher;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.reindex.RestReindexAction.ReindexParseContext;
 import org.elasticsearch.index.reindex.remote.RemoteInfo;
 import org.elasticsearch.indices.query.IndicesQueriesRegistry;
+import org.elasticsearch.rest.RestController;
 import org.elasticsearch.search.SearchRequestParsers;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.rest.FakeRestRequest;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
+import static java.util.Collections.singletonMap;
 import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
+import static org.mockito.Mockito.mock;
 
 public class RestReindexActionTests extends ESTestCase {
     public void testBuildRemoteInfoNoRemote() throws IOException {
@@ -127,6 +132,31 @@ public class RestReindexActionTests extends ESTestCase {
         }
     }
 
+    public void testPipelineQueryParameterIsError() throws IOException {
+        SearchRequestParsers parsers = new SearchRequestParsers(new IndicesQueriesRegistry(), null, null, null);
+        RestReindexAction action = new RestReindexAction(Settings.EMPTY, mock(RestController.class), parsers, null);
+
+        FakeRestRequest.Builder request = new FakeRestRequest.Builder();
+        try (XContentBuilder body = JsonXContent.contentBuilder().prettyPrint()) {
+            body.startObject(); {
+                body.startObject("source"); {
+                    body.field("index", "source");
+                }
+                body.endObject();
+                body.startObject("dest"); {
+                    body.field("index", "dest");
+                }
+                body.endObject();
+            }
+            body.endObject();
+            request.withContent(body.bytes());
+        }
+        request.withParams(singletonMap("pipeline", "doesn't matter"));
+        Exception e = expectThrows(IllegalArgumentException.class, () -> action.buildRequest(request.build()));
+
+        assertEquals("_reindex doesn't support [pipeline] as a query parmaeter. Specify it in the [dest] object instead.", e.getMessage());
+    }
+
     private RemoteInfo buildRemoteInfoHostTestCase(String hostInRest) throws IOException {
         Map<String, Object> remote = new HashMap<>();
         remote.put("host", hostInRest);