Reindex in mixed cluster

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

@@ -64,6 +64,7 @@ public class ReindexPlugin extends Plugin implements ActionPlugin, PersistentTas
 
     private final SetOnce<ScriptService> scriptService = new SetOnce<>();
     private final SetOnce<ReindexSslConfig> reindexSslConfig = new SetOnce<>();
+    private final SetOnce<ClusterService> clusterService = new SetOnce<>();
 
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
@@ -98,7 +99,7 @@ public List<RestHandler> getRestHandlers(Settings settings, RestController restC
             IndexScopedSettings indexScopedSettings, SettingsFilter settingsFilter, IndexNameExpressionResolver indexNameExpressionResolver,
             Supplier<DiscoveryNodes> nodesInCluster) {
         return Arrays.asList(
-                new RestReindexAction(restController),
+                new RestReindexAction(restController, clusterService.get()),
                 new RestUpdateByQueryAction(restController),
                 new RestDeleteByQueryAction(restController),
                 new RestRethrottleAction(restController, nodesInCluster));
@@ -112,6 +113,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
         this.scriptService.set(scriptService);
         this.reindexSslConfig.set(new ReindexSslConfig(environment.settings(), environment, resourceWatcherService));
         namedXContentRegistry.set(xContentRegistry);
+        this.clusterService.set(clusterService);
         return Collections.singletonList(reindexSslConfig.get());
     }
 

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

@@ -19,9 +19,11 @@
 
 package org.elasticsearch.index.reindex;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BytesRestResponse;
@@ -43,8 +45,11 @@
  */
 public class RestReindexAction extends AbstractBaseReindexRestHandler<ReindexRequest, ReindexAction> {
 
-    public RestReindexAction(RestController controller) {
+    private final ClusterService clusterService;
+
+    public RestReindexAction(RestController controller, ClusterService clusterService) {
         super(ReindexAction.INSTANCE);
+        this.clusterService = clusterService;
         controller.registerHandler(POST, "/_reindex", this);
     }
 
@@ -55,8 +60,10 @@ public String getName() {
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        if (System.getProperty("es.reindex.resilience", "true").equals("false")) {
-            // todo: remove this escape hatch in 8.0
+        // todo: remove system property escape hatch in 8.0
+        // todo: fix version constant on backport to 7.x
+        if (clusterService.state().nodes().getMinNodeVersion().before(Version.V_8_0_0)
+                || System.getProperty("es.reindex.resilience", "true").equals("false")) {
             return doPrepareRequest(request, client, true, true);
         }
 

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

@@ -38,7 +38,7 @@ public class RestReindexActionTests extends RestActionTestCase {
 
     @Before
     public void setUpAction() {
-        action = new RestReindexAction(controller());
+        action = new RestReindexAction(controller(), null);
     }
 
     public void testPipelineQueryParameterIsError() throws IOException {

qa/mixed-cluster/src/test/resources/rest-api-spec/test/mixed_cluster/reindex/10_basic.yml

@@ -0,0 +1,36 @@
+---
+"Verify that reindex completes successfully in a mixed cluster":
+  - do:
+      indices.create:
+        index: reindex_mixed_cluster_source
+        body:
+          settings:
+            index:
+              number_of_replicas: 0
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "reindex_mixed_cluster_source"}}'
+          - '{"f1": "1"}'
+          - '{"index": {"_index": "reindex_mixed_cluster_source"}}'
+          - '{"f1": "2"}'
+          - '{"index": {"_index": "reindex_mixed_cluster_source"}}'
+          - '{"f1": "3"}'
+          - '{"index": {"_index": "reindex_mixed_cluster_source"}}'
+          - '{"f1": "4"}'
+          - '{"index": {"_index": "reindex_mixed_cluster_source"}}'
+          - '{"f1": "5"}'
+
+  - do:
+      reindex:
+        body:
+          source:
+            index: reindex_mixed_cluster_source
+            size: 1
+          dest:
+            index: reindex_mixed_custer_copy
+  - match: {created: 5}
+  - match: {version_conflicts: 0}
+  - match: {batches: 5}
+  - match: {failures: []}