Use CcrRepository to init follower index

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/restore/RestoreClusterStateListener.java

@@ -0,0 +1,90 @@
+/*
+ * 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.admin.cluster.snapshots.restore;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.RestoreInProgress;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.snapshots.RestoreInfo;
+import org.elasticsearch.snapshots.RestoreService;
+import org.elasticsearch.snapshots.Snapshot;
+
+import static org.elasticsearch.snapshots.RestoreService.restoreInProgress;
+
+public class RestoreClusterStateListener implements ClusterStateListener {
+
+    private static final Logger logger = LogManager.getLogger(RestoreClusterStateListener.class);
+
+    private final ClusterService clusterService;
+    private final String uuid;
+    private final Snapshot snapshot;
+    private final ActionListener<RestoreSnapshotResponse> listener;
+
+
+    private RestoreClusterStateListener(ClusterService clusterService, RestoreService.RestoreCompletionResponse response,
+                                        ActionListener<RestoreSnapshotResponse> listener) {
+        this.clusterService = clusterService;
+        this.uuid = response.getUuid();
+        this.snapshot = response.getSnapshot();
+        this.listener = listener;
+    }
+
+    @Override
+    public void clusterChanged(ClusterChangedEvent changedEvent) {
+        final RestoreInProgress.Entry prevEntry = restoreInProgress(changedEvent.previousState(), uuid);
+        final RestoreInProgress.Entry newEntry = restoreInProgress(changedEvent.state(), uuid);
+        if (prevEntry == null) {
+            // When there is a master failure after a restore has been started, this listener might not be registered
+            // on the current master and as such it might miss some intermediary cluster states due to batching.
+            // Clean up listener in that case and acknowledge completion of restore operation to client.
+            clusterService.removeListener(this);
+            listener.onResponse(new RestoreSnapshotResponse(null));
+        } else if (newEntry == null) {
+            clusterService.removeListener(this);
+            ImmutableOpenMap<ShardId, RestoreInProgress.ShardRestoreStatus> shards = prevEntry.shards();
+            assert prevEntry.state().completed() : "expected completed snapshot state but was " + prevEntry.state();
+            assert RestoreService.completed(shards) : "expected all restore entries to be completed";
+            RestoreInfo ri = new RestoreInfo(prevEntry.snapshot().getSnapshotId().getName(),
+                prevEntry.indices(),
+                shards.size(),
+                shards.size() - RestoreService.failedShards(shards));
+            RestoreSnapshotResponse response = new RestoreSnapshotResponse(ri);
+            logger.debug("restore of [{}] completed", snapshot);
+            listener.onResponse(response);
+        } else {
+            // restore not completed yet, wait for next cluster state update
+        }
+    }
+
+    /**
+     * Creates a cluster state listener and registers it with the cluster service. The listener passed as a
+     * parameter will be called with the restore is complete.
+     */
+    public static void createAndRegisterListener(ClusterService clusterService, RestoreService.RestoreCompletionResponse response,
+                                                 ActionListener<RestoreSnapshotResponse> listener) {
+        clusterService.addListener(new RestoreClusterStateListener(clusterService, response, listener));
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/restore/TransportRestoreSnapshotAction.java

@@ -22,26 +22,17 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
-import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateListener;
-import org.elasticsearch.cluster.RestoreInProgress;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.index.shard.ShardId;
-import org.elasticsearch.snapshots.RestoreInfo;
 import org.elasticsearch.snapshots.RestoreService;
 import org.elasticsearch.snapshots.RestoreService.RestoreCompletionResponse;
-import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import static org.elasticsearch.snapshots.RestoreService.restoreInProgress;
-
 /**
  * Transport action for restore snapshot operation
  */
@@ -91,39 +82,7 @@ protected void masterOperation(final RestoreSnapshotRequest request, final Clust
             @Override
             public void onResponse(RestoreCompletionResponse restoreCompletionResponse) {
                 if (restoreCompletionResponse.getRestoreInfo() == null && request.waitForCompletion()) {
-                    final Snapshot snapshot = restoreCompletionResponse.getSnapshot();
-                    String uuid = restoreCompletionResponse.getUuid();
-
-                    ClusterStateListener clusterStateListener = new ClusterStateListener() {
-                        @Override
-                        public void clusterChanged(ClusterChangedEvent changedEvent) {
-                            final RestoreInProgress.Entry prevEntry = restoreInProgress(changedEvent.previousState(), uuid);
-                            final RestoreInProgress.Entry newEntry = restoreInProgress(changedEvent.state(), uuid);
-                            if (prevEntry == null) {
-                                // When there is a master failure after a restore has been started, this listener might not be registered
-                                // on the current master and as such it might miss some intermediary cluster states due to batching.
-                                // Clean up listener in that case and acknowledge completion of restore operation to client.
-                                clusterService.removeListener(this);
-                                listener.onResponse(new RestoreSnapshotResponse(null));
-                            } else if (newEntry == null) {
-                                clusterService.removeListener(this);
-                                ImmutableOpenMap<ShardId, RestoreInProgress.ShardRestoreStatus> shards = prevEntry.shards();
-                                assert prevEntry.state().completed() : "expected completed snapshot state but was " + prevEntry.state();
-                                assert RestoreService.completed(shards) : "expected all restore entries to be completed";
-                                RestoreInfo ri = new RestoreInfo(prevEntry.snapshot().getSnapshotId().getName(),
-                                    prevEntry.indices(),
-                                    shards.size(),
-                                    shards.size() - RestoreService.failedShards(shards));
-                                RestoreSnapshotResponse response = new RestoreSnapshotResponse(ri);
-                                logger.debug("restore of [{}] completed", snapshot);
-                                listener.onResponse(response);
-                            } else {
-                                // restore not completed yet, wait for next cluster state update
-                            }
-                        }
-                    };
-
-                    clusterService.addListener(clusterStateListener);
+                    RestoreClusterStateListener.createAndRegisterListener(clusterService, restoreCompletionResponse, listener);
                 } else {
                     listener.onResponse(new RestoreSnapshotResponse(restoreCompletionResponse.getRestoreInfo()));
                 }

x-pack/plugin/ccr/qa/rest/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml

@@ -38,6 +38,7 @@
       ccr.follow:
         index: bar
         body:
+          wait_for_completion: true
           remote_cluster: local
           leader_index: foo
   - is_true: follow_index_created

x-pack/plugin/ccr/qa/rest/src/test/resources/rest-api-spec/test/ccr/follow_stats.yml

@@ -37,6 +37,7 @@
       ccr.follow:
         index: bar
         body:
+          wait_for_completion: true
           remote_cluster: local
           leader_index: foo
   - is_true: follow_index_created

x-pack/plugin/ccr/qa/rest/src/test/resources/rest-api-spec/test/ccr/index_directly_into_follower_index.yml

@@ -38,6 +38,7 @@
       ccr.follow:
         index: bar
         body:
+          wait_for_completion: true
           remote_cluster: local
           leader_index: foo
   - is_true: follow_index_created

x-pack/plugin/ccr/qa/src/main/java/org/elasticsearch/xpack/ccr/ESCCRRestTestCase.java

@@ -75,7 +75,7 @@ protected static void followIndex(String leaderCluster, String leaderIndex, Stri
     protected static void followIndex(RestClient client, String leaderCluster, String leaderIndex, String followIndex) throws IOException {
         final Request request = new Request("PUT", "/" + followIndex + "/_ccr/follow");
         request.setJsonEntity("{\"remote_cluster\": \"" + leaderCluster + "\", \"leader_index\": \"" + leaderIndex +
-            "\", \"read_poll_timeout\": \"10ms\"}");
+            "\", \"read_poll_timeout\": \"10ms\", \"wait_for_completion\": true}");
         assertOK(client.performRequest(request));
     }
 
@@ -186,6 +186,7 @@ protected static Map<String, Object> toMap(String response) {
     protected static void ensureYellow(String index) throws IOException {
         Request request = new Request("GET", "/_cluster/health/" + index);
         request.addParameter("wait_for_status", "yellow");
+        request.addParameter("wait_for_active_shards", "1");
         request.addParameter("wait_for_no_relocating_shards", "true");
         request.addParameter("wait_for_no_initializing_shards", "true");
         request.addParameter("timeout", "70s");

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java

@@ -532,6 +532,7 @@ private void followLeaderIndex(String autoFollowPattenName,
             request.setRemoteCluster(remoteCluster);
             request.setLeaderIndex(indexToFollow.getName());
             request.setFollowRequest(followRequest);
+            request.setWaitForCompletion(true);
 
             // Execute if the create and follow api call succeeds:
             Runnable successHandler = () -> {