Avoid NPE in follower stats when no tasks metadata (#35802)

When there is no persistent tasks metadata we could hit a null pointer
exception when executing a follower stats request. This is because we
inspect the persistent tasks metadata. Yet, if no tasks have been
registered, this is null (as opposed to empty). We need to avoid
de-referencing the persistent tasks metadata in this case. That is what
this commit does, and we add a test for this situation.

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

@@ -84,6 +84,11 @@ public class TransportFollowStatsAction extends TransportTasksAction<
     protected void processTasks(final FollowStatsAction.StatsRequest request, final Consumer<ShardFollowNodeTask> operation) {
         final ClusterState state = clusterService.state();
         final PersistentTasksCustomMetaData persistentTasksMetaData = state.metaData().custom(PersistentTasksCustomMetaData.TYPE);
+
+        if (persistentTasksMetaData == null) {
+            return;
+        }
+
         final Set<String> followerIndices = persistentTasksMetaData.tasks().stream()
             .filter(persistentTask -> persistentTask.getTaskName().equals(ShardFollowTask.NAME))
             .map(persistentTask -> {

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.xpack.CcrSingleNodeTestCase;
+import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+
+/*
+ * Test scope is important to ensure that other tests added to this suite do not interfere with the expectation in
+ * testStatsWhenNoPersistentTasksMetaDataExists that the cluster state does not contain any persistent tasks metadata.
+ */
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
+public class FollowStatsIT extends CcrSingleNodeTestCase {
+
+    /**
+     * Previously we would throw a NullPointerException when there was no persistent tasks metadata in the cluster state. This tests
+     * maintains that we do not make this mistake again.
+     *
+     * @throws InterruptedException if we are interrupted waiting on the latch to countdown
+     */
+    public void testStatsWhenNoPersistentTasksMetaDataExists() throws InterruptedException {
+        final ClusterStateResponse response = client().admin().cluster().state(new ClusterStateRequest()).actionGet();
+        assertNull(response.getState().metaData().custom(PersistentTasksCustomMetaData.TYPE));
+        final AtomicBoolean onResponse = new AtomicBoolean();
+        final CountDownLatch latch = new CountDownLatch(1);
+        client().execute(
+                FollowStatsAction.INSTANCE,
+                new FollowStatsAction.StatsRequest(),
+                new ActionListener<FollowStatsAction.StatsResponses>() {
+                    @Override
+                    public void onResponse(final FollowStatsAction.StatsResponses statsResponses) {
+                        try {
+                            assertThat(statsResponses.getTaskFailures(), empty());
+                            assertThat(statsResponses.getNodeFailures(), empty());
+                            onResponse.set(true);
+                        } finally {
+                            latch.countDown();
+                        }
+                    }
+
+                    @Override
+                    public void onFailure(final Exception e) {
+                        try {
+                            fail(e.toString());
+                        } finally {
+                            latch.countDown();
+                        }
+                    }
+                });
+        latch.await();
+        assertTrue(onResponse.get());
+    }
+
+}