Fix shard follow task startup error handling (#39053)

Prior to this commit, if during fetch leader / follower GCP
a fatal error occurred, then the shard follow task was removed.

This is unexpected, because if such an error occurs during the lifetime of shard follow task then replication is stopped and the fatal error flag is set. This allows the ccr stats api to report the fatal exception that has occurred (instead of the user grepping through the elasticsearch logs).

This issue was found by a rare failure of the  `FollowStatsIT#testFollowStatsApiIncludeShardFollowStatsWithRemovedFollowerIndex` test.

Closes #38779

Author: martijnvg

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

@@ -452,11 +452,15 @@ private void handleFailure(Exception e, AtomicInteger retryCounter, Runnable tas
                 scheduler.accept(TimeValue.timeValueMillis(delay), task);
             }
         } else {
-            fatalException = ExceptionsHelper.convertToElastic(e);
-            LOGGER.warn("shard follow task encounter non-retryable error", e);
+            setFatalException(e);
         }
     }
 
+    void setFatalException(Exception e) {
+        fatalException = ExceptionsHelper.convertToElastic(e);
+        LOGGER.warn("shard follow task encounter non-retryable error", e);
+    }
+
     static long computeDelay(int currentRetry, long maxRetryDelayInMillis) {
         // Cap currentRetry to avoid overflow when computing n variable
         int maxCurrentRetry = Math.min(currentRetry, 24);

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

@@ -282,7 +282,7 @@ protected void nodeOperation(final AllocatedPersistentTask task, final ShardFoll
                     shardFollowNodeTask), e);
                 threadPool.schedule(() -> nodeOperation(task, params, state), params.getMaxRetryDelay(), Ccr.CCR_THREAD_POOL_NAME);
             } else {
-                shardFollowNodeTask.markAsFailed(e);
+                shardFollowNodeTask.setFatalException(e);
             }
         };
 

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

@@ -149,7 +149,6 @@ public void testFollowStatsApiResourceNotFound() throws Exception {
         assertAcked(client().execute(PauseFollowAction.INSTANCE, new PauseFollowAction.Request("follower1")).actionGet());
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/38779")
     public void testFollowStatsApiIncludeShardFollowStatsWithRemovedFollowerIndex() throws Exception {
         final String leaderIndexSettings = getIndexSettings(1, 0,
             singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));