Fixed CCR stats api serialization issues and (#33983)

always use `IndicesOptions.strictExpand()` for indices options.

The follow index may be closed and we still want to get stats from
shard follow task and the whether the provided index name matches with
follow index name is checked when locating the task itself in the ccr
stats transport action.

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

@@ -6,7 +6,6 @@
 
 package org.elasticsearch.xpack.ccr.rest;
 
-import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
@@ -35,7 +34,6 @@ public class RestCcrStatsAction extends BaseRestHandler {
     protected RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
         final CcrStatsAction.StatsRequest request = new CcrStatsAction.StatsRequest();
         request.setIndices(Strings.splitStringByCommaToArray(restRequest.param("index")));
-        request.setIndicesOptions(IndicesOptions.fromRequest(restRequest, request.indicesOptions()));
         return channel -> client.execute(CcrStatsAction.INSTANCE, request, new RestToXContentListener<>(channel));
     }
 

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

@@ -0,0 +1,26 @@
+/*
+ * 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.test.AbstractStreamableTestCase;
+import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
+
+public class StatsRequestTests extends AbstractStreamableTestCase<CcrStatsAction.StatsRequest> {
+
+    @Override
+    protected CcrStatsAction.StatsRequest createBlankInstance() {
+        return new CcrStatsAction.StatsRequest();
+    }
+
+    @Override
+    protected CcrStatsAction.StatsRequest createTestInstance() {
+        CcrStatsAction.StatsRequest statsRequest = new CcrStatsAction.StatsRequest();
+        if (randomBoolean()) {
+            statsRequest.setIndices(generateRandomStringArray(8, 4, false));
+        }
+        return statsRequest;
+    }
+}

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

@@ -0,0 +1,56 @@
+/*
+ * 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.test.AbstractStreamableTestCase;
+import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
+import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class StatsResponsesTests extends AbstractStreamableTestCase<CcrStatsAction.StatsResponses> {
+
+    @Override
+    protected CcrStatsAction.StatsResponses createBlankInstance() {
+        return new CcrStatsAction.StatsResponses();
+    }
+
+    @Override
+    protected CcrStatsAction.StatsResponses createTestInstance() {
+        int numResponses = randomIntBetween(0, 8);
+        List<CcrStatsAction.StatsResponse> responses = new ArrayList<>(numResponses);
+        for (int i = 0; i < numResponses; i++) {
+            ShardFollowNodeTaskStatus status = new ShardFollowNodeTaskStatus(
+                randomAlphaOfLength(4),
+                randomAlphaOfLength(4),
+                randomInt(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomIntBetween(0, Integer.MAX_VALUE),
+                randomIntBetween(0, Integer.MAX_VALUE),
+                randomIntBetween(0, Integer.MAX_VALUE),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                Collections.emptyNavigableMap(),
+                randomLong());
+            responses.add(new CcrStatsAction.StatsResponse(status));
+        }
+        return new CcrStatsAction.StatsResponses(Collections.emptyList(), Collections.emptyList(), responses);
+    }
+}

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

@@ -23,9 +23,11 @@ import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.TreeMap;
 
 public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
@@ -45,7 +47,7 @@ public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
 
     public static class StatsResponses extends BaseTasksResponse implements ToXContentObject {
 
-        private final List<StatsResponse> statsResponse;
+        private List<StatsResponse> statsResponse;
 
         public List<StatsResponse> getStatsResponses() {
             return statsResponse;
@@ -87,6 +89,31 @@ public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
             builder.endObject();
             return builder;
         }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            super.readFrom(in);
+            statsResponse = in.readList(StatsResponse::new);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeList(statsResponse);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            StatsResponses that = (StatsResponses) o;
+            return Objects.equals(statsResponse, that.statsResponse);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(statsResponse);
+        }
     }
 
     public static class StatsRequest extends BaseTasksRequest<StatsRequest> implements IndicesRequest {
@@ -102,15 +129,9 @@ public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
             this.indices = indices;
         }
 
-        private IndicesOptions indicesOptions = IndicesOptions.strictExpandOpenAndForbidClosed();
-
         @Override
         public IndicesOptions indicesOptions() {
-            return indicesOptions;
+            return IndicesOptions.strictExpand();
-        }
-
-        public void setIndicesOptions(final IndicesOptions indicesOptions) {
-            this.indicesOptions = indicesOptions;
         }
 
         @Override
@@ -134,17 +155,27 @@ public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
         @Override
         public void readFrom(final StreamInput in) throws IOException {
             super.readFrom(in);
-            indices = in.readStringArray();
+            indices = in.readOptionalStringArray();
-            indicesOptions = IndicesOptions.readIndicesOptions(in);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
-            out.writeStringArray(indices);
+            out.writeOptionalStringArray(indices);
-            indicesOptions.writeIndicesOptions(out);
         }
 
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            StatsRequest that = (StatsRequest) o;
+            return Arrays.equals(indices, that.indices);
+        }
+
+        @Override
+        public int hashCode() {
+            return Arrays.hashCode(indices);
+        }
     }
 
     public static class StatsResponse implements Writeable {
@@ -168,6 +199,18 @@ public class CcrStatsAction extends Action<CcrStatsAction.StatsResponses> {
             status.writeTo(out);
         }
 
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            StatsResponse that = (StatsResponse) o;
+            return Objects.equals(status, that.status);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(status);
+        }
     }
 
 }

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/ccr/CcrStatsCollector.java

@@ -6,7 +6,6 @@
 
 package org.elasticsearch.xpack.monitoring.collector.ccr;
 
-import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
@@ -51,10 +50,10 @@ public final class CcrStatsCollector extends AbstractCcrCollector {
         long interval,
         MonitoringDoc.Node node) throws Exception {
 
-        final CcrStatsAction.StatsRequest request = new CcrStatsAction.StatsRequest();
+
-        request.setIndices(getCollectionIndices());
+            final CcrStatsAction.StatsRequest request = new CcrStatsAction.StatsRequest();
-        request.setIndicesOptions(IndicesOptions.lenientExpandOpen());
+            request.setIndices(getCollectionIndices());
-        final CcrStatsAction.StatsResponses responses = ccrClient.stats(request).actionGet(getCollectionTimeout());
+            final CcrStatsAction.StatsResponses responses = ccrClient.stats(request).actionGet(getCollectionTimeout());
 
         return responses
             .getStatsResponses()