Use CcrRepository to init follower index

client/rest-high-level/src/main/java/org/elasticsearch/client/CcrRequestConverters.java

@@ -46,6 +46,8 @@ static Request putFollow(PutFollowRequest putFollowRequest) throws IOException {
             .addPathPartAsIs("_ccr", "follow")
             .build();
         Request request = new Request(HttpPut.METHOD_NAME, endpoint);
+        RequestConverters.Params parameters = new RequestConverters.Params(request);
+        parameters.withWaitForActiveShards(putFollowRequest.waitForActiveShards());
         request.setEntity(createEntity(putFollowRequest, REQUEST_BODY_CONTENT_TYPE));
         return request;
     }

client/rest-high-level/src/main/java/org/elasticsearch/client/ccr/PutFollowRequest.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.client.ccr;
 
+import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.Validatable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.xcontent.ToXContentObject;
@@ -36,11 +37,17 @@ public final class PutFollowRequest extends FollowConfig implements Validatable,
     private final String remoteCluster;
     private final String leaderIndex;
     private final String followerIndex;
+    private final ActiveShardCount waitForActiveShards;
 
     public PutFollowRequest(String remoteCluster, String leaderIndex, String followerIndex) {
+        this(remoteCluster, leaderIndex, followerIndex, ActiveShardCount.NONE);
+    }
+
+    public PutFollowRequest(String remoteCluster, String leaderIndex, String followerIndex, ActiveShardCount waitForActiveShards) {
         this.remoteCluster = Objects.requireNonNull(remoteCluster, "remoteCluster");
         this.leaderIndex = Objects.requireNonNull(leaderIndex, "leaderIndex");
         this.followerIndex = Objects.requireNonNull(followerIndex, "followerIndex");
+        this.waitForActiveShards = waitForActiveShards;
     }
 
     @Override
@@ -66,13 +73,18 @@ public String getFollowerIndex() {
         return followerIndex;
     }
 
+    public ActiveShardCount waitForActiveShards() {
+        return waitForActiveShards;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         if (!super.equals(o)) return false;
         PutFollowRequest that = (PutFollowRequest) o;
-        return Objects.equals(remoteCluster, that.remoteCluster) &&
+        return Objects.equals(waitForActiveShards, that.waitForActiveShards) &&
+            Objects.equals(remoteCluster, that.remoteCluster) &&
             Objects.equals(leaderIndex, that.leaderIndex) &&
             Objects.equals(followerIndex, that.followerIndex);
     }
@@ -83,7 +95,7 @@ public int hashCode() {
             super.hashCode(),
             remoteCluster,
             leaderIndex,
-            followerIndex
-        );
+            followerIndex,
+            waitForActiveShards);
     }
 }

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

@@ -27,6 +27,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.ccr.CcrStatsRequest;
 import org.elasticsearch.client.ccr.CcrStatsResponse;
@@ -95,7 +96,7 @@ public void testIndexFollowing() throws Exception {
         CreateIndexResponse response = highLevelClient().indices().create(createIndexRequest, RequestOptions.DEFAULT);
         assertThat(response.isAcknowledged(), is(true));
 
-        PutFollowRequest putFollowRequest = new PutFollowRequest("local_cluster", "leader", "follower");
+        PutFollowRequest putFollowRequest = new PutFollowRequest("local_cluster", "leader", "follower", ActiveShardCount.ONE);
         PutFollowResponse putFollowResponse = execute(putFollowRequest, ccrClient::putFollow, ccrClient::putFollowAsync);
         assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
         assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.ESRestHighLevelClientTestCase;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
@@ -97,7 +98,8 @@ public void testPutFollow() throws Exception {
         PutFollowRequest putFollowRequest = new PutFollowRequest(
             "local", // <1>
             "leader", // <2>
-            "follower" // <3>
+            "follower", // <3>
+            ActiveShardCount.ONE // <4>
         );
         // end::ccr-put-follow-request
 
@@ -175,7 +177,7 @@ public void testPauseFollow() throws Exception {
         String followIndex = "follower";
         // Follow index, so that it can be paused:
         {
-            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex);
+            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex, ActiveShardCount.ONE);
             PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT);
             assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
             assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));
@@ -241,7 +243,7 @@ public void testResumeFollow() throws Exception {
         String followIndex = "follower";
         // Follow index, so that it can be paused:
         {
-            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex);
+            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex, ActiveShardCount.ONE);
             PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT);
             assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
             assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));
@@ -317,7 +319,7 @@ public void testUnfollow() throws Exception {
         String followIndex = "follower";
         // Follow index, pause and close, so that it can be unfollowed:
         {
-            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex);
+            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex, ActiveShardCount.ONE);
             PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT);
             assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
             assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));
@@ -349,7 +351,7 @@ public void testUnfollow() throws Exception {
             DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(followIndex);
             assertThat(client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT).isAcknowledged(), is(true));
 
-            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex);
+            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex, ActiveShardCount.ONE);
             PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT);
             assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
             assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));
@@ -639,7 +641,7 @@ public void testGetFollowStats() throws Exception {
         }
         {
             // Follow index, so that we can query for follow stats:
-            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", "follower");
+            PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", "follower", ActiveShardCount.ONE);
             PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT);
             assertThat(putFollowResponse.isFollowIndexCreated(), is(true));
             assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true));

docs/java-rest/high-level/ccr/put_follow.asciidoc

@@ -20,6 +20,8 @@ include-tagged::{doc-tests-file}[{api}-request]
 <1> The name of the remote cluster alias.
 <2> The name of the leader in the remote cluster.
 <3> The name of the follower index that gets created as part of the put follow API call.
+<4> The number of active shard copies to wait for before the put follow API returns a
+response, as an `ActiveShardCount`
 
 [id="{upid}-{api}-response"]
 ==== Response

docs/reference/ccr/apis/follow/get-follow-info.asciidoc

@@ -22,7 +22,7 @@ replication options and whether the follower indices are active or paused.
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/apis/follow/get-follow-stats.asciidoc

@@ -21,7 +21,7 @@ following tasks associated with each shard for the specified indices.
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/apis/follow/post-pause-follow.asciidoc

@@ -24,7 +24,7 @@ following task.
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/apis/follow/post-resume-follow.asciidoc

@@ -23,7 +23,7 @@ returns, the follower index will resume fetching operations from the leader inde
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/apis/follow/post-unfollow.asciidoc

@@ -27,7 +27,7 @@ irreversible operation.
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/apis/follow/put-follow.asciidoc

@@ -31,7 +31,7 @@ POST /follower_index/_ccr/pause_follow
 
 [source,js]
 --------------------------------------------------
-PUT /<follower_index>/_ccr/follow
+PUT /<follower_index>/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "<remote_cluster>",
   "leader_index" : "<leader_index>"
@@ -43,6 +43,11 @@ PUT /<follower_index>/_ccr/follow
 // TEST[s/<remote_cluster>/remote_cluster/]
 // TEST[s/<leader_index>/leader_index/]
 
+The `wait_for_active_shards` parameter specifies the number of shards to wait on being active
+before responding. This defaults to waiting on none of the shards to be active. A shard must
+be restored from the leader index being active. Restoring a follower shard requires transferring
+all the remote lucene segment files to the follower index.
+
 ==== Path Parameters
 
 `follower_index` (required)::
@@ -73,7 +78,7 @@ This example creates a follower index named `follower_index`:
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index",

docs/reference/ccr/apis/get-ccr-stats.asciidoc

@@ -22,7 +22,7 @@ shard-level stats as in the <<ccr-get-follow-stats,get follower stats API>>.
 
 [source,js]
 --------------------------------------------------
-PUT /follower_index/_ccr/follow
+PUT /follower_index/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "remote_cluster",
   "leader_index" : "leader_index"

docs/reference/ccr/getting-started.asciidoc

@@ -230,7 +230,7 @@ cluster.
 
 [source,js]
 --------------------------------------------------
-PUT /server-metrics-copy/_ccr/follow
+PUT /server-metrics-copy/_ccr/follow?wait_for_active_shards=1
 {
   "remote_cluster" : "leader",
   "leader_index" : "server-metrics"

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

@@ -0,0 +1,87 @@
+/*
+ * 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 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 ActionListener<RestoreSnapshotResponse> listener;
+
+
+    private RestoreClusterStateListener(ClusterService clusterService, RestoreService.RestoreCompletionResponse response,
+                                        ActionListener<RestoreSnapshotResponse> listener) {
+        this.clusterService = clusterService;
+        this.uuid = response.getUuid();
+        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", prevEntry.snapshot().getSnapshotId());
+            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 when the restore is complete.
+     */
+    public static void createAndRegisterListener(ClusterService clusterService, RestoreService.RestoreCompletionResponse response,
+                                                 ActionListener<RestoreSnapshotResponse> listener) {
+        clusterService.addListener(new RestoreClusterStateListener(clusterService, response, listener));
+    }
+}