Rewrite shard follow node task logic

server/src/main/java/org/elasticsearch/action/support/replication/TransportWriteAction.java

@@ -130,13 +130,15 @@ public static class WritePrimaryResult<ReplicaRequest extends ReplicatedWriteReq
             implements RespondingWriteResult {
         boolean finishedAsyncActions;
         public final Location location;
+        public final IndexShard primary;
         ActionListener<Response> listener = null;
 
         public WritePrimaryResult(ReplicaRequest request, @Nullable Response finalResponse,
                                   @Nullable Location location, @Nullable Exception operationFailure,
                                   IndexShard primary, Logger logger) {
             super(request, finalResponse, operationFailure);
             this.location = location;
+            this.primary = primary;
             assert location == null || operationFailure == null
                     : "expected either failure to be null or translog location to be null, " +
                     "but found: [" + location + "] translog location and [" + operationFailure + "] failure";

server/src/main/java/org/elasticsearch/index/engine/LuceneChangesSnapshot.java

@@ -141,14 +141,14 @@ public Translog.Operation next() throws IOException {
     private void rangeCheck(Translog.Operation op) {
         if (op == null) {
             if (lastSeenSeqNo < toSeqNo) {
-                throw new IllegalStateException("Not all operations between min_seqno [" + fromSeqNo + "] " +
-                    "and max_seqno [" + toSeqNo + "] found; prematurely terminated last_seen_seqno [" + lastSeenSeqNo + "]");
+                throw new IllegalStateException("Not all operations between from_seqno [" + fromSeqNo + "] " +
+                    "and to_seqno [" + toSeqNo + "] found; prematurely terminated last_seen_seqno [" + lastSeenSeqNo + "]");
             }
         } else {
             final long expectedSeqNo = lastSeenSeqNo + 1;
             if (op.seqNo() != expectedSeqNo) {
-                throw new IllegalStateException("Not all operations between min_seqno [" + fromSeqNo + "] " +
-                    "and max_seqno [" + toSeqNo + "] found; expected seqno [" + expectedSeqNo + "]; found [" + op + "]");
+                throw new IllegalStateException("Not all operations between from_seqno [" + fromSeqNo + "] " +
+                    "and to_seqno [" + toSeqNo + "] found; expected seqno [" + expectedSeqNo + "]; found [" + op + "]");
             }
         }
     }

server/src/test/java/org/elasticsearch/index/engine/LuceneChangesSnapshotTests.java

@@ -65,7 +65,7 @@ public void testBasics() throws Exception {
         try (Translog.Snapshot snapshot = engine.newLuceneChangesSnapshot("test", mapperService, fromSeqNo, toSeqNo, true)) {
             IllegalStateException error = expectThrows(IllegalStateException.class, () -> drainAll(snapshot));
             assertThat(error.getMessage(),
-                containsString("Not all operations between min_seqno [" + fromSeqNo + "] and max_seqno [" + toSeqNo + "] found"));
+                containsString("Not all operations between from_seqno [" + fromSeqNo + "] and to_seqno [" + toSeqNo + "] found"));
         }
         try (Translog.Snapshot snapshot = engine.newLuceneChangesSnapshot("test", mapperService, fromSeqNo, toSeqNo, false)) {
             assertThat(snapshot, SnapshotMatchers.size(0));
@@ -108,19 +108,19 @@ searcher, mapperService, between(1, LuceneChangesSnapshot.DEFAULT_BATCH_SIZE), f
                 searcher = null;
                 IllegalStateException error = expectThrows(IllegalStateException.class, () -> drainAll(snapshot));
                 assertThat(error.getMessage(),
-                    containsString("Not all operations between min_seqno [" + fromSeqNo + "] and max_seqno [" + toSeqNo + "] found"));
+                    containsString("Not all operations between from_seqno [" + fromSeqNo + "] and to_seqno [" + toSeqNo + "] found"));
             }finally {
                 IOUtils.close(searcher);
             }
-        }else {
+        } else {
             fromSeqNo = randomLongBetween(0, refreshedSeqNo);
             toSeqNo = randomLongBetween(refreshedSeqNo + 1, numOps * 2);
             Engine.Searcher searcher = engine.acquireSearcher("test", Engine.SearcherScope.INTERNAL);
             try (Translog.Snapshot snapshot = new LuceneChangesSnapshot(
                 searcher, mapperService, between(1, LuceneChangesSnapshot.DEFAULT_BATCH_SIZE), fromSeqNo, toSeqNo, false)) {
                 searcher = null;
                 assertThat(snapshot, SnapshotMatchers.containsSeqNoRange(fromSeqNo, refreshedSeqNo));
-            }finally {
+            } finally {
                 IOUtils.close(searcher);
             }
             searcher = engine.acquireSearcher("test", Engine.SearcherScope.INTERNAL);
@@ -129,7 +129,7 @@ searcher, mapperService, between(1, LuceneChangesSnapshot.DEFAULT_BATCH_SIZE), f
                 searcher = null;
                 IllegalStateException error = expectThrows(IllegalStateException.class, () -> drainAll(snapshot));
                 assertThat(error.getMessage(),
-                    containsString("Not all operations between min_seqno [" + fromSeqNo + "] and max_seqno [" + toSeqNo + "] found"));
+                    containsString("Not all operations between from_seqno [" + fromSeqNo + "] and to_seqno [" + toSeqNo + "] found"));
             }finally {
                 IOUtils.close(searcher);
             }
@@ -139,7 +139,7 @@ searcher, mapperService, between(1, LuceneChangesSnapshot.DEFAULT_BATCH_SIZE), f
                 searcher, mapperService, between(1, LuceneChangesSnapshot.DEFAULT_BATCH_SIZE), fromSeqNo, toSeqNo, true)) {
                 searcher = null;
                 assertThat(snapshot, SnapshotMatchers.containsSeqNoRange(fromSeqNo, toSeqNo));
-            }finally {
+            } finally {
                 IOUtils.close(searcher);
             }
         }

x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java

@@ -84,7 +84,7 @@ public void testFollowIndex() throws Exception {
                 assertThat(tasks.size(), equalTo(0));
                 assertThat(countCcrNodeTasks(), equalTo(0));
             });
-    
+
             followIndex("leader_cluster:" + allowedIndex, allowedIndex);
             assertThat(countCcrNodeTasks(), equalTo(1));
             assertOK(client().performRequest(new Request("POST", "/" + allowedIndex + "/_xpack/ccr/_unfollow")));
@@ -95,14 +95,14 @@ public void testFollowIndex() throws Exception {
                 assertThat(tasks.size(), equalTo(0));
                 assertThat(countCcrNodeTasks(), equalTo(0));
             });
-    
+
             createAndFollowIndex("leader_cluster:" + unallowedIndex, unallowedIndex);
             // Verify that nothing has been replicated and no node tasks are running
             // These node tasks should have been failed due to the fact that the user
             // has no sufficient priviledges.
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
             verifyDocuments(adminClient(), unallowedIndex, 0);
-            
+
             followIndex("leader_cluster:" + unallowedIndex, unallowedIndex);
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
             verifyDocuments(adminClient(), unallowedIndex, 0);
@@ -146,12 +146,14 @@ private static void refresh(String index) throws IOException {
     private static void followIndex(String leaderIndex, String followIndex) throws IOException {
         final Request request = new Request("POST", "/" + followIndex + "/_xpack/ccr/_follow");
         request.addParameter("leader_index", leaderIndex);
+        request.addParameter("idle_shard_retry_delay", "10ms");
         assertOK(client().performRequest(request));
     }
 
     private static void createAndFollowIndex(String leaderIndex, String followIndex) throws IOException {
         final Request request = new Request("POST", "/" + followIndex + "/_xpack/ccr/_create_and_follow");
         request.addParameter("leader_index", leaderIndex);
+        request.addParameter("idle_shard_retry_delay", "10ms");
         assertOK(client().performRequest(request));
     }
 

x-pack/plugin/ccr/qa/multi-cluster/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexIT.java

@@ -96,12 +96,14 @@ private static void refresh(String index) throws IOException {
     private static void followIndex(String leaderIndex, String followIndex) throws IOException {
         final Request request = new Request("POST", "/" + followIndex + "/_xpack/ccr/_follow");
         request.addParameter("leader_index", leaderIndex);
+        request.addParameter("idle_shard_retry_delay", "10ms");
         assertOK(client().performRequest(request));
     }
 
     private static void createAndFollowIndex(String leaderIndex, String followIndex) throws IOException {
         final Request request = new Request("POST", "/" + followIndex + "/_xpack/ccr/_create_and_follow");
         request.addParameter("leader_index", leaderIndex);
+        request.addParameter("idle_shard_retry_delay", "10ms");
         assertOK(client().performRequest(request));
     }
 

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

@@ -65,12 +65,15 @@ public static class Request extends AcknowledgedRequest<Request> {
 
         private FollowIndexAction.Request followRequest;
 
-        public FollowIndexAction.Request getFollowRequest() {
-            return followRequest;
+        public Request(FollowIndexAction.Request followRequest) {
+            this.followRequest = Objects.requireNonNull(followRequest);
+        }
+
+        Request() {
         }
 
-        public void setFollowRequest(FollowIndexAction.Request followRequest) {
-            this.followRequest = followRequest;
+        public FollowIndexAction.Request getFollowRequest() {
+            return followRequest;
         }
 
         @Override

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexingSlowLog;
 import org.elasticsearch.index.SearchSlowLog;
@@ -71,50 +72,57 @@ public static class Request extends ActionRequest {
 
         private String leaderIndex;
         private String followIndex;
-        private long batchSize = ShardFollowTasksExecutor.DEFAULT_BATCH_SIZE;
-        private int concurrentProcessors = ShardFollowTasksExecutor.DEFAULT_CONCURRENT_PROCESSORS;
-        private long processorMaxTranslogBytes = ShardFollowTasksExecutor.DEFAULT_MAX_TRANSLOG_BYTES;
-
-        public String getLeaderIndex() {
-            return leaderIndex;
-        }
-
-        public void setLeaderIndex(String leaderIndex) {
-            this.leaderIndex = leaderIndex;
-        }
-
-        public String getFollowIndex() {
-            return followIndex;
-        }
+        private int maxBatchOperationCount;
+        private int maxConcurrentReadBatches;
+        private long maxOperationSizeInBytes;
+        private int maxConcurrentWriteBatches;
+        private int maxWriteBufferSize;
+        private TimeValue retryTimeout;
+        private TimeValue idleShardRetryDelay;
+
+        public Request(String leaderIndex, String followIndex, int maxBatchOperationCount, int maxConcurrentReadBatches,
+                       long maxOperationSizeInBytes, int maxConcurrentWriteBatches, int maxWriteBufferSize,
+                       TimeValue retryTimeout, TimeValue idleShardRetryDelay) {
+            if (maxBatchOperationCount < 1) {
+                throw new IllegalArgumentException("maxBatchOperationCount must be larger than 0");
+            }
+            if (maxConcurrentReadBatches < 1) {
+                throw new IllegalArgumentException("concurrent_processors must be larger than 0");
+            }
+            if (maxOperationSizeInBytes <= 0) {
+                throw new IllegalArgumentException("processor_max_translog_bytes must be larger than 0");
+            }
+            if (maxConcurrentWriteBatches < 1) {
+                throw new IllegalArgumentException("maxConcurrentWriteBatches must be larger than 0");
+            }
+            if (maxWriteBufferSize < 1) {
+                throw new IllegalArgumentException("maxWriteBufferSize must be larger than 0");
+            }
 
-        public void setFollowIndex(String followIndex) {
-            this.followIndex = followIndex;
+            this.leaderIndex = Objects.requireNonNull(leaderIndex);
+            this.followIndex = Objects.requireNonNull(followIndex);
+            this.maxBatchOperationCount = maxBatchOperationCount;
+            this.maxConcurrentReadBatches = maxConcurrentReadBatches;
+            this.maxOperationSizeInBytes = maxOperationSizeInBytes;
+            this.maxConcurrentWriteBatches = maxConcurrentWriteBatches;
+            this.maxWriteBufferSize = maxWriteBufferSize;
+            this.retryTimeout = retryTimeout;
+            this.idleShardRetryDelay = idleShardRetryDelay;
         }
 
-        public long getBatchSize() {
-            return batchSize;
+        Request() {
         }
 
-        public void setBatchSize(long batchSize) {
-            if (batchSize < 1) {
-                throw new IllegalArgumentException("Illegal batch_size [" + batchSize + "]");
-            }
-
-            this.batchSize = batchSize;
+        public String getLeaderIndex() {
+            return leaderIndex;
         }
 
-        public void setConcurrentProcessors(int concurrentProcessors) {
-            if (concurrentProcessors < 1) {
-                throw new IllegalArgumentException("concurrent_processors must be larger than 0");
-            }
-            this.concurrentProcessors = concurrentProcessors;
+        public String getFollowIndex() {
+            return followIndex;
         }
 
-        public void setProcessorMaxTranslogBytes(long processorMaxTranslogBytes) {
-            if (processorMaxTranslogBytes <= 0) {
-                throw new IllegalArgumentException("processor_max_translog_bytes must be larger than 0");
-            }
-            this.processorMaxTranslogBytes = processorMaxTranslogBytes;
+        public int getMaxBatchOperationCount() {
+            return maxBatchOperationCount;
         }
 
         @Override
@@ -127,36 +135,49 @@ public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             leaderIndex = in.readString();
             followIndex = in.readString();
-            batchSize = in.readVLong();
-            concurrentProcessors = in.readVInt();
-            processorMaxTranslogBytes = in.readVLong();
+            maxBatchOperationCount = in.readVInt();
+            maxConcurrentReadBatches = in.readVInt();
+            maxOperationSizeInBytes = in.readVLong();
+            maxConcurrentWriteBatches = in.readVInt();
+            maxWriteBufferSize = in.readVInt();
+            retryTimeout = in.readOptionalTimeValue();
+            idleShardRetryDelay = in.readOptionalTimeValue();
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(leaderIndex);
             out.writeString(followIndex);
-            out.writeVLong(batchSize);
-            out.writeVInt(concurrentProcessors);
-            out.writeVLong(processorMaxTranslogBytes);
+            out.writeVInt(maxBatchOperationCount);
+            out.writeVInt(maxConcurrentReadBatches);
+            out.writeVLong(maxOperationSizeInBytes);
+            out.writeVInt(maxConcurrentWriteBatches);
+            out.writeVInt(maxWriteBufferSize);
+            out.writeOptionalTimeValue(retryTimeout);
+            out.writeOptionalTimeValue(idleShardRetryDelay);
         }
 
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return batchSize == request.batchSize &&
-                concurrentProcessors == request.concurrentProcessors &&
-                processorMaxTranslogBytes == request.processorMaxTranslogBytes &&
+            return maxBatchOperationCount == request.maxBatchOperationCount &&
+                maxConcurrentReadBatches == request.maxConcurrentReadBatches &&
+                maxOperationSizeInBytes == request.maxOperationSizeInBytes &&
+                maxConcurrentWriteBatches == request.maxConcurrentWriteBatches &&
+                maxWriteBufferSize == request.maxWriteBufferSize &&
+                Objects.equals(retryTimeout, request.retryTimeout) &&
+                Objects.equals(idleShardRetryDelay, request.idleShardRetryDelay) &&
                 Objects.equals(leaderIndex, request.leaderIndex) &&
                 Objects.equals(followIndex, request.followIndex);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(leaderIndex, followIndex, batchSize, concurrentProcessors, processorMaxTranslogBytes);
+            return Objects.hash(leaderIndex, followIndex, maxBatchOperationCount, maxConcurrentReadBatches, maxOperationSizeInBytes,
+                maxConcurrentWriteBatches, maxWriteBufferSize, retryTimeout, idleShardRetryDelay);
         }
     }
 
@@ -197,7 +218,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
             ClusterState localClusterState = clusterService.state();
             IndexMetaData followIndexMetadata = localClusterState.getMetaData().index(request.followIndex);
 
-            String[] indices = new String[]{request.getLeaderIndex()};
+            String[] indices = new String[]{request.leaderIndex};
             Map<String, List<String>> remoteClusterIndices = remoteClusterService.groupClusterIndices(indices, s -> false);
             if (remoteClusterIndices.containsKey(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
                 // Following an index in local cluster, so use local cluster state to fetch leader IndexMetaData:
@@ -251,10 +272,21 @@ void start(Request request, String clusterNameAlias, IndexMetaData leaderIndexMe
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));for (int i = 0; i < numShards; i++) {
                 final int shardId = i;
                 String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
+                TimeValue retryTimeout = ShardFollowNodeTask.DEFAULT_RETRY_TIMEOUT;
+                if (request.retryTimeout != null) {
+                    retryTimeout = request.retryTimeout;
+                }
+                TimeValue idleShardRetryDelay = ShardFollowNodeTask.DEFAULT_IDLE_SHARD_RETRY_DELAY;
+                if (request.idleShardRetryDelay != null) {
+                    idleShardRetryDelay = request.idleShardRetryDelay;
+                }
+
                 ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
                         new ShardId(followIndexMetadata.getIndex(), shardId),
                         new ShardId(leaderIndexMetadata.getIndex(), shardId),
-                        request.batchSize, request.concurrentProcessors, request.processorMaxTranslogBytes, filteredHeaders);
+                        request.maxBatchOperationCount, request.maxConcurrentReadBatches, request.maxOperationSizeInBytes,
+                        request.maxConcurrentWriteBatches, request.maxWriteBufferSize, retryTimeout, idleShardRetryDelay,
+                        filteredHeaders);
                 persistentTasksService.sendStartRequest(taskId, ShardFollowTask.NAME, shardFollowTask,
                         new ActionListener<PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask>>() {
                             @Override