Prepare ShardFollowNodeTask to bootstrap when it fall behind leader shard

* Changed the shard changes api to include a special metadata in the exception being thrown
to indicate that the ops are no longer there.
* Changed ShardFollowNodeTask to handle this exception with special metadata
  and mark a shard as fallen behind its leader shard. The shard follow task
  will then abort its on going replication.

The code that does the restore from ccr repository still needs to be added.
This change should make that change a bit easier.

Relates to elastic#35975

Author: martijnvg

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

@@ -113,6 +113,8 @@ public class Ccr extends Plugin implements ActionPlugin, PersistentTaskPlugin, E
     public static final String CCR_CUSTOM_METADATA_LEADER_INDEX_NAME_KEY = "leader_index_name";
     public static final String CCR_CUSTOM_METADATA_REMOTE_CLUSTER_NAME_KEY = "remote_cluster_name";
 
+    public static final String FALLEN_BEHIND_LEADER_SHARD_METADATA_KEY = "es.fallen_behind_leader_shard";
+
     private final boolean enabled;
     private final Settings settings;
     private final CcrLicenseChecker ccrLicenseChecker;

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

@@ -37,6 +37,7 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.Ccr;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -397,7 +398,11 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
                 if (cause instanceof IllegalStateException && cause.getMessage().contains("Not all operations between from_seqno [")) {
                     String message = "Operations are no longer available for replicating. Maybe increase the retention setting [" +
                         IndexSettings.INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING.getKey() + "]?";
-                    listener.onFailure(new ElasticsearchException(message, e));
+                    // Make it easy to detect this error in ShardFollowNodeTask:
+                    // (adding a metadata header instead of introducing a new exception that extends ElasticsearchException)
+                    ElasticsearchException wrapper = new ElasticsearchException(message, e);
+                    wrapper.addMetadata(Ccr.FALLEN_BEHIND_LEADER_SHARD_METADATA_KEY);
+                    listener.onFailure(wrapper);
                 } else {
                     listener.onFailure(e);
                 }

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

@@ -30,6 +30,7 @@
 import org.elasticsearch.persistent.AllocatedPersistentTask;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.transport.ConnectTransportException;
+import org.elasticsearch.xpack.ccr.Ccr;
 import org.elasticsearch.xpack.ccr.action.bulk.BulkShardOperationsResponse;
 import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
 
@@ -43,6 +44,7 @@
 import java.util.Queue;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -90,6 +92,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
     private final LinkedHashMap<Long, Tuple<AtomicInteger, ElasticsearchException>> fetchExceptions;
 
     private volatile ElasticsearchException fatalException;
+    private final AtomicBoolean fallenBehindLeaderShard = new AtomicBoolean(false);
 
     ShardFollowNodeTask(long id, String type, String action, String description, TaskId parentTask, Map<String, String> headers,
                         ShardFollowTask params, BiConsumer<TimeValue, Runnable> scheduler, final LongSupplier relativeTimeProvider) {
@@ -274,6 +277,13 @@ private void sendShardChangesRequest(long from, int maxOperationCount, long maxR
                         failedReadRequests++;
                         fetchExceptions.put(from, Tuple.tuple(retryCounter, ExceptionsHelper.convertToElastic(e)));
                     }
+                    if (e instanceof ElasticsearchException) {
+                        ElasticsearchException elasticsearchException = (ElasticsearchException) e;
+                        if (elasticsearchException.getMetadataKeys().contains(Ccr.FALLEN_BEHIND_LEADER_SHARD_METADATA_KEY)) {
+                            handleFallenBehindLeaderShard(e);
+                            return;
+                        }
+                    }
                     handleFailure(e, retryCounter, () -> sendShardChangesRequest(from, maxOperationCount, maxRequiredSeqNo, retryCounter));
                 });
     }
@@ -290,6 +300,17 @@ void handleReadResponse(long from, long maxRequiredSeqNo, ShardChangesAction.Res
         maybeUpdateSettings(response.getSettingsVersion(), updateMappingsTask);
     }
 
+    void handleFallenBehindLeaderShard(Exception e) {
+        if (fallenBehindLeaderShard.compareAndSet(false, true)) {
+            LOGGER.warn(new ParameterizedMessage("{} shard follow task has fallen behind the leader shard {} that it is following",
+                params.getFollowShardId(), params.getLeaderShardId()), e);
+
+            // Do restore from repository here and
+            // after start() should be invoked and
+            // stats should be reset including fallenBehindLeaderShard
+        }
+    }
+
     /** Called when some operations are fetched from the leading */
     protected void onOperationsFetched(Translog.Operation[] operations) {
 
@@ -487,7 +508,7 @@ protected void onCancelled() {
     }
 
     protected boolean isStopped() {
-        return fatalException != null || isCancelled() || isCompleted();
+        return fallenBehindLeaderShard.get() || fatalException != null || isCancelled() || isCompleted();
     }
 
     public ShardId getFollowShardId() {
@@ -536,7 +557,8 @@ public synchronized ShardFollowNodeTaskStatus getStatus() {
                                 .collect(
                                         Collectors.toMap(Map.Entry::getKey, e -> Tuple.tuple(e.getValue().v1().get(), e.getValue().v2())))),
                 timeSinceLastFetchMillis,
-                fatalException);
+                fatalException,
+                fallenBehindLeaderShard.get());
     }
 
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardChangesTests.java

@@ -16,12 +16,14 @@
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.xpack.ccr.Ccr;
 import org.elasticsearch.xpack.ccr.LocalStateCcr;
 
 import java.util.Collection;
 import java.util.Collections;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
 
 public class ShardChangesTests extends ESSingleNodeTestCase {
 
@@ -113,9 +115,12 @@ public void testMissingOperations() {
         request.setFromSeqNo(0L);
         request.setMaxOperationCount(1);
 
-        Exception e = expectThrows(ElasticsearchException.class, () -> client().execute(ShardChangesAction.INSTANCE, request).actionGet());
+        ElasticsearchException e =
+            expectThrows(ElasticsearchException.class, () -> client().execute(ShardChangesAction.INSTANCE, request).actionGet());
         assertThat(e.getMessage(), equalTo("Operations are no longer available for replicating. Maybe increase the retention setting " +
             "[index.soft_deletes.retention.operations]?"));
+        assertThat(e.getMetadataKeys().size(), equalTo(1));
+        assertThat(e.getMetadata(Ccr.FALLEN_BEHIND_LEADER_SHARD_METADATA_KEY), notNullValue());
     }
 
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTaskStatusTests.java

@@ -61,7 +61,8 @@ protected ShardFollowNodeTaskStatus createTestInstance() {
                 randomNonNegativeLong(),
                 randomReadExceptions(),
                 randomLong(),
-                randomBoolean() ? new ElasticsearchException("fatal error") : null);
+                randomBoolean() ? new ElasticsearchException("fatal error") : null,
+                randomBoolean());
     }
 
     @Override

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTaskTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.index.shard.ShardNotFoundException;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.ccr.Ccr;
 import org.elasticsearch.xpack.ccr.action.bulk.BulkShardOperationsResponse;
 import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
 
@@ -42,6 +43,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.sameInstance;
+import static org.hamcrest.core.Is.is;
 
 public class ShardFollowNodeTaskTests extends ESTestCase {
 
@@ -287,6 +289,37 @@ public void testReceiveRetryableError() {
         assertThat(status.leaderGlobalCheckpoint(), equalTo(63L));
     }
 
+    public void testFallenBehindLeaderShard() {
+        ShardFollowTaskParams params = new ShardFollowTaskParams();
+        params.maxReadRequestOperationCount = 64;
+        params.maxOutstandingReadRequests = 1;
+        params.maxOutstandingWriteRequests = 1;
+        ShardFollowNodeTask task = createShardFollowTask(params);
+        startTask(task, 63, -1);
+
+        ElasticsearchException exception = new ElasticsearchException("no ops for you");
+        exception.addMetadata(Ccr.FALLEN_BEHIND_LEADER_SHARD_METADATA_KEY);
+        readFailures.add(exception);
+        mappingVersions.add(1L);
+        leaderGlobalCheckpoints.add(63L);
+        maxSeqNos.add(63L);
+        responseSizes.add(64);
+        simulateResponse.set(true);
+        task.coordinateReads();
+
+        assertThat(task.isStopped(), is(true));
+        ShardFollowNodeTaskStatus status = task.getStatus();
+        assertThat(status.fallenBehindLeaderShard(), is(true));
+        assertThat(status.outstandingReadRequests(), equalTo(1));
+        assertThat(status.outstandingWriteRequests(), equalTo(0));
+        assertThat(status.failedReadRequests(), equalTo(1L));
+        assertThat(status.successfulReadRequests(), equalTo(0L));
+        assertThat(status.readExceptions().size(), equalTo(1));
+        assertThat(status.readExceptions().values().iterator().next().v2(), sameInstance(exception));
+        assertThat(status.lastRequestedSeqNo(), equalTo(63L));
+        assertThat(status.leaderGlobalCheckpoint(), equalTo(63L));
+    }
+
     public void testEmptyShardChangesResponseShouldClearFetchException() {
         ShardFollowTaskParams params = new ShardFollowTaskParams();
         params.maxReadRequestOperationCount = 64;

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/StatsResponsesTests.java

@@ -59,7 +59,8 @@ static FollowStatsAction.StatsResponses createStatsResponse() {
                 randomNonNegativeLong(),
                 Collections.emptyNavigableMap(),
                 randomLong(),
-                randomBoolean() ? new ElasticsearchException("fatal error") : null);
+                randomBoolean() ? new ElasticsearchException("fatal error") : null,
+                randomBoolean());
             responses.add(new FollowStatsAction.StatsResponse(status));
         }
         return new FollowStatsAction.StatsResponses(Collections.emptyList(), Collections.emptyList(), responses);

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/monitoring/collector/ccr/FollowStatsMonitoringDocTests.java

@@ -110,6 +110,7 @@ public void testToXContent() throws IOException {
                         randomNonNegativeLong(),
                         Tuple.tuple(randomIntBetween(0, Integer.MAX_VALUE), new ElasticsearchException("shard is sad"))));
         final long timeSinceLastReadMillis = randomNonNegativeLong();
+        final boolean fallenBehindLeaderShard = randomBoolean();
         final ShardFollowNodeTaskStatus status = new ShardFollowNodeTaskStatus(
                 "leader_cluster",
                 "leader_index",
@@ -138,7 +139,8 @@ public void testToXContent() throws IOException {
                 operationWritten,
                 fetchExceptions,
                 timeSinceLastReadMillis,
-                new ElasticsearchException("fatal error"));
+                new ElasticsearchException("fatal error"),
+                fallenBehindLeaderShard);
         final FollowStatsMonitoringDoc document = new FollowStatsMonitoringDoc("_cluster", timestamp, intervalMillis, node, status);
         final BytesReference xContent = XContentHelper.toXContent(document, XContentType.JSON, false);
         assertThat(
@@ -194,7 +196,8 @@ public void testToXContent() throws IOException {
                                                 + "}"
                                         + "],"
                                         + "\"time_since_last_read_millis\":" + timeSinceLastReadMillis + ","
-                                        + "\"fatal_exception\":{\"type\":\"exception\",\"reason\":\"fatal error\"}"
+                                        + "\"fatal_exception\":{\"type\":\"exception\",\"reason\":\"fatal error\"},"
+                                        + "\"fallen_behind_leader_shard\":" + fallenBehindLeaderShard
                                 + "}"
                         + "}"));
     }
@@ -230,7 +233,8 @@ public void testShardFollowNodeTaskStatusFieldsMapped() throws IOException {
             10,
             fetchExceptions,
             2,
-            null);
+            null,
+            false);
         XContentBuilder builder = jsonBuilder();
         builder.value(status);
         Map<String, Object> serializedStatus = XContentHelper.convertToMap(XContentType.JSON.xContent(), Strings.toString(builder), false);
@@ -252,6 +256,8 @@ public void testShardFollowNodeTaskStatusFieldsMapped() throws IOException {
             } else if (fieldValue instanceof String) {
                 assertThat("expected keyword field type for field [" + fieldName + "]", fieldType,
                     anyOf(equalTo("keyword"), equalTo("text")));
+            } else if (fieldValue instanceof Boolean) {
+                assertThat("expected keyword field type for field [" + fieldName + "]", fieldType, equalTo("boolean"));
             } else {
                 // Manual test specific object fields and if not just fail:
                 if (fieldName.equals("read_exceptions")) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/ShardFollowNodeTaskStatus.java

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.core.ccr;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Tuple;
@@ -62,6 +63,7 @@ public class ShardFollowNodeTaskStatus implements Task.Status {
     private static final ParseField READ_EXCEPTIONS = new ParseField("read_exceptions");
     private static final ParseField TIME_SINCE_LAST_READ_MILLIS_FIELD = new ParseField("time_since_last_read_millis");
     private static final ParseField FATAL_EXCEPTION = new ParseField("fatal_exception");
+    private static final ParseField FALLEN_BEHIND_LEADER_SHARD = new ParseField("fallen_behind_leader_shard");
 
     @SuppressWarnings("unchecked")
     static final ConstructingObjectParser<ShardFollowNodeTaskStatus, Void> STATUS_PARSER =
@@ -98,7 +100,8 @@ public class ShardFollowNodeTaskStatus implements Task.Status {
                                             .stream()
                                             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))),
                             (long) args[26],
-                            (ElasticsearchException) args[27]));
+                            (ElasticsearchException) args[27],
+                            (boolean) args[28]));
 
     public static final String READ_EXCEPTIONS_ENTRY_PARSER_NAME = "shard-follow-node-task-status-read-exceptions-entry";
 
@@ -138,6 +141,7 @@ public class ShardFollowNodeTaskStatus implements Task.Status {
         STATUS_PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(),
                 (p, c) -> ElasticsearchException.fromXContent(p),
                 FATAL_EXCEPTION);
+        STATUS_PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), FALLEN_BEHIND_LEADER_SHARD);
     }
 
     static final ParseField READ_EXCEPTIONS_ENTRY_FROM_SEQ_NO = new ParseField("from_seq_no");
@@ -321,6 +325,12 @@ public ElasticsearchException getFatalException() {
         return fatalException;
     }
 
+    private final boolean fallenBehindLeaderShard;
+
+    public boolean fallenBehindLeaderShard() {
+        return fallenBehindLeaderShard;
+    }
+
     public ShardFollowNodeTaskStatus(
             final String remoteCluster,
             final String leaderIndex,
@@ -349,7 +359,8 @@ public ShardFollowNodeTaskStatus(
             final long operationWritten,
             final NavigableMap<Long, Tuple<Integer, ElasticsearchException>> readExceptions,
             final long timeSinceLastReadMillis,
-            final ElasticsearchException fatalException) {
+            final ElasticsearchException fatalException,
+            final boolean fallenBehindLeaderShard) {
         this.remoteCluster = remoteCluster;
         this.leaderIndex = leaderIndex;
         this.followerIndex = followerIndex;
@@ -378,6 +389,7 @@ public ShardFollowNodeTaskStatus(
         this.readExceptions = Objects.requireNonNull(readExceptions);
         this.timeSinceLastReadMillis = timeSinceLastReadMillis;
         this.fatalException = fatalException;
+        this.fallenBehindLeaderShard = fallenBehindLeaderShard;
     }
 
     public ShardFollowNodeTaskStatus(final StreamInput in) throws IOException {
@@ -410,6 +422,11 @@ public ShardFollowNodeTaskStatus(final StreamInput in) throws IOException {
                 new TreeMap<>(in.readMap(StreamInput::readVLong, stream -> Tuple.tuple(stream.readVInt(), stream.readException())));
         this.timeSinceLastReadMillis = in.readZLong();
         this.fatalException = in.readException();
+        if (in.getVersion().onOrAfter(Version.V_6_7_0)) {
+            this.fallenBehindLeaderShard = in.readBoolean();
+        } else {
+            this.fallenBehindLeaderShard = false;
+        }
     }
 
     @Override
@@ -453,6 +470,9 @@ public void writeTo(final StreamOutput out) throws IOException {
                 });
         out.writeZLong(timeSinceLastReadMillis);
         out.writeException(fatalException);
+        if (out.getVersion().onOrAfter(Version.V_6_7_0)) {
+            out.writeBoolean(fallenBehindLeaderShard);
+        }
     }
 
     @Override
@@ -536,6 +556,7 @@ public XContentBuilder toXContentFragment(final XContentBuilder builder, final P
             }
             builder.endObject();
         }
+        builder.field(FALLEN_BEHIND_LEADER_SHARD.getPreferredName(), fallenBehindLeaderShard);
         return builder;
     }
 
@@ -582,7 +603,8 @@ public boolean equals(final Object o) {
                 readExceptions.keySet().equals(that.readExceptions.keySet()) &&
                 getReadExceptionMessages(this).equals(getReadExceptionMessages(that)) &&
                 timeSinceLastReadMillis == that.timeSinceLastReadMillis &&
-                Objects.equals(fatalExceptionMessage, otherFatalExceptionMessage);
+                Objects.equals(fatalExceptionMessage, otherFatalExceptionMessage) &&
+                Objects.equals(fallenBehindLeaderShard, that.fallenBehindLeaderShard);
     }
 
     @Override
@@ -620,7 +642,8 @@ public int hashCode() {
                 readExceptions.keySet(),
                 getReadExceptionMessages(this),
                 timeSinceLastReadMillis,
-                fatalExceptionMessage);
+                fatalExceptionMessage,
+                fallenBehindLeaderShard);
     }
 
     private static List<String> getReadExceptionMessages(final ShardFollowNodeTaskStatus status) {

x-pack/plugin/core/src/main/resources/monitoring-es.json

@@ -1028,6 +1028,9 @@
             },
             "time_since_last_read_millis": {
               "type": "long"
+            },
+            "fallen_behind_leader_shard": {
+              "type": "boolean"
             }
           }
         },