Fix serialization

Author: nielsbauman

server/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateRequest.java

@@ -9,14 +9,12 @@
 
 package org.elasticsearch.action.admin.cluster.state;
 
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.local.LocalClusterStateRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.tasks.CancellableTask;
@@ -63,25 +61,6 @@ public ClusterStateRequest(StreamInput in) throws IOException {
         waitForMetadataVersion = in.readOptionalLong();
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        getParentTask().writeTo(out);
-        out.writeTimeValue(masterTimeout());
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
-            out.writeVLong(0L); // Master term
-        } // else no protection against routing loops in older versions
-        out.writeBoolean(true); // Local
-        out.writeBoolean(routingTable);
-        out.writeBoolean(nodes);
-        out.writeBoolean(metadata);
-        out.writeBoolean(blocks);
-        out.writeBoolean(customs);
-        out.writeStringArray(indices);
-        indicesOptions.writeIndicesOptions(out);
-        out.writeTimeValue(waitForTimeout);
-        out.writeOptionalLong(waitForMetadataVersion);
-    }
-
     @Override
     public ActionRequestValidationException validate() {
         return null;

server/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateResponse.java

@@ -15,7 +15,6 @@
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.core.UpdateForV10;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -64,11 +63,6 @@ public boolean isWaitForTimedOut() {
         return waitForTimedOut;
     }
 
-    /**
-     * NB prior to 9.0 this was a TransportMasterNodeReadAction so for BwC we must remain able to write these responses until
-     * we no longer need to support calling this action remotely.
-     */
-    @UpdateForV10(owner = UpdateForV10.Owner.DATA_MANAGEMENT)
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         clusterName.writeTo(out);

server/src/main/java/org/elasticsearch/action/admin/cluster/state/RemoteClusterStateRequest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.cluster.state;
+
+import org.elasticsearch.TransportVersions;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class RemoteClusterStateRequest extends ActionRequest {
+
+    private final ClusterStateRequest clusterStateRequest;
+
+    public RemoteClusterStateRequest(ClusterStateRequest clusterStateRequest) {
+        this.clusterStateRequest = clusterStateRequest;
+    }
+
+    public RemoteClusterStateRequest(StreamInput in) throws IOException {
+        this.clusterStateRequest = new ClusterStateRequest(in);
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return clusterStateRequest.validate();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        getParentTask().writeTo(out);
+        out.writeTimeValue(clusterStateRequest.masterTimeout());
+        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
+            out.writeVLong(0L); // Master term
+        } // else no protection against routing loops in older versions
+        out.writeBoolean(true); // Local
+        out.writeBoolean(clusterStateRequest.routingTable());
+        out.writeBoolean(clusterStateRequest.nodes());
+        out.writeBoolean(clusterStateRequest.metadata());
+        out.writeBoolean(clusterStateRequest.blocks());
+        out.writeBoolean(clusterStateRequest.customs());
+        out.writeStringArray(clusterStateRequest.indices());
+        clusterStateRequest.indicesOptions().writeIndicesOptions(out);
+        out.writeTimeValue(clusterStateRequest.waitForTimeout());
+        out.writeOptionalLong(clusterStateRequest.waitForMetadataVersion());
+    }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+    }
+
+    @Override
+    public String getDescription() {
+        return clusterStateRequest.getDescription();
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateAction.java

@@ -14,7 +14,6 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.ChannelActionListener;
 import org.elasticsearch.action.support.local.TransportLocalClusterStateAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
@@ -59,11 +58,6 @@ public class TransportClusterStateAction extends TransportLocalClusterStateActio
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final ThreadPool threadPool;
 
-    /**
-     * NB prior to 9.0 this was a TransportMasterNodeReadAction so for BwC it must be registered with the TransportService until
-     * we no longer need to support calling this action remotely.
-     */
-    @SuppressWarnings("this-escape")
     @Inject
     public TransportClusterStateAction(
         TransportService transportService,
@@ -83,15 +77,6 @@ public TransportClusterStateAction(
         this.projectResolver = projectResolver;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.threadPool = threadPool;
-
-        transportService.registerRequestHandler(
-            actionName,
-            executor,
-            false,
-            true,
-            ClusterStateRequest::new,
-            (request, channel, task) -> executeDirect(task, request, new ChannelActionListener<>(channel))
-        );
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportRemoteClusterStateAction.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.cluster.state;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+public class TransportRemoteClusterStateAction extends HandledTransportAction<RemoteClusterStateRequest, ClusterStateResponse> {
+
+    private final Client client;
+
+    @Inject
+    protected TransportRemoteClusterStateAction(
+        TransportService transportService,
+        ThreadPool threadPool,
+        ActionFilters actionFilters,
+        Client client
+    ) {
+        super(
+            ClusterStateAction.NAME,
+            transportService,
+            actionFilters,
+            RemoteClusterStateRequest::new,
+            threadPool.executor(ThreadPool.Names.MANAGEMENT)
+        );
+        this.client = client;
+    }
+
+    @Override
+    protected void doExecute(Task task, RemoteClusterStateRequest request, ActionListener<ClusterStateResponse> listener) {
+        client.execute(ClusterStateAction.INSTANCE, request, listener);
+    }
+}

server/src/main/java/org/elasticsearch/action/support/local/LocalClusterStateRequest.java

@@ -64,7 +64,7 @@ protected LocalClusterStateRequest(StreamInput in, boolean readLocal) throws IOE
     }
 
     @Override
-    public void writeTo(StreamOutput out) throws IOException {
+    public final void writeTo(StreamOutput out) throws IOException {
         TransportAction.localOnly();
     }
 

server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.cluster.state.RemoteClusterStateRequest;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.Settings;
@@ -158,7 +159,7 @@ void collectNodes(ActionListener<Function<String, DiscoveryNode>> listener) {
                     transportService.sendRequest(
                         connection,
                         ClusterStateAction.NAME,
-                        request,
+                        new RemoteClusterStateRequest(request),
                         TransportRequestOptions.EMPTY,
                         new ActionListenerResponseHandler<>(
                             contextPreservingActionListener.map(response -> response.getState().nodes()::get),

server/src/main/java/org/elasticsearch/transport/SniffConnectionStrategy.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.cluster.state.RemoteClusterStateRequest;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
@@ -324,7 +325,7 @@ private void collectRemoteNodes(Iterator<Supplier<DiscoveryNode>> seedNodesSuppl
                     final ClusterStateRequest clusterStateRequest = new ClusterStateRequest(SNIFF_REQUEST_TIMEOUT);
                     clusterStateRequest.clear();
                     clusterStateRequest.nodes(true);
-                    request = clusterStateRequest;
+                    request = new RemoteClusterStateRequest(clusterStateRequest);
                     sniffResponseHandler = new ClusterStateSniffResponseHandler(connection, listener, seedNodesSuppliers);
                 }
 

server/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java

@@ -14,8 +14,8 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.remote.RemoteClusterNodesAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
-import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.cluster.state.RemoteClusterStateRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchShardsRequest;
 import org.elasticsearch.action.search.SearchShardsResponse;
@@ -163,7 +163,7 @@ public static MockTransportService startTransport(
             newService.registerRequestHandler(
                 ClusterStateAction.NAME,
                 EsExecutors.DIRECT_EXECUTOR_SERVICE,
-                ClusterStateRequest::new,
+                RemoteClusterStateRequest::new,
                 (request, channel, task) -> {
                     DiscoveryNodes.Builder builder = DiscoveryNodes.builder();
                     for (DiscoveryNode node : knownNodes) {

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

@@ -14,8 +14,8 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.RemoteClusterActionType;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
-import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.cluster.state.RemoteClusterStateRequest;
 import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
 import org.elasticsearch.action.admin.indices.stats.IndexStats;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsAction;
@@ -134,7 +134,7 @@ public void checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
             client,
             clusterAlias,
             remoteClient,
-            CcrRequests.metadataRequest(leaderIndex),
+            new RemoteClusterStateRequest(CcrRequests.metadataRequest(leaderIndex)),
             onFailure,
             remoteClusterStateResponse -> {
                 ClusterState remoteClusterState = remoteClusterStateResponse.getState();
@@ -201,7 +201,7 @@ public void checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
     public static void checkRemoteClusterLicenseAndFetchClusterState(
         final Client client,
         final String clusterAlias,
-        final ClusterStateRequest request,
+        final RemoteClusterStateRequest request,
         final Consumer<Exception> onFailure,
         final Consumer<ClusterStateResponse> leaderClusterStateConsumer
     ) {
@@ -250,7 +250,7 @@ private static void checkRemoteClusterLicenseAndFetchClusterState(
         final Client client,
         final String clusterAlias,
         final RemoteClusterClient remoteClient,
-        final ClusterStateRequest request,
+        final RemoteClusterStateRequest request,
         final Consumer<Exception> onFailure,
         final Consumer<ClusterStateResponse> leaderClusterStateConsumer,
         final Function<RemoteClusterLicenseChecker.LicenseCheck, ElasticsearchStatusException> nonCompliantLicense,