Change CCR API request classes to use Writeable serialization instead of Streamable (elastic#34911)

Only the follow stats request couldn't be changed to use Writeable serialization, because that requires changes in `TransportTasksAction` and `BaseTasksRequest` base classes.

Author: martijnvg

server/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java

@@ -41,6 +41,11 @@ public abstract class AcknowledgedRequest<Request extends MasterNodeRequest<Requ
  protected AcknowledgedRequest() {
  }
 
+ protected AcknowledgedRequest(StreamInput in) throws IOException {
+ super(in);
+ timeout = in.readTimeValue();
+ }
+
  /**
  * Allows to set the timeout
  * @param timeout timeout as a string (e.g. 1s)

server/src/main/java/org/elasticsearch/action/support/master/MasterNodeReadRequest.java

@@ -31,6 +31,14 @@ public abstract class MasterNodeReadRequest<Request extends MasterNodeReadReques
 
  protected boolean local = false;
 
+ protected MasterNodeReadRequest() {
+ }
+
+ protected MasterNodeReadRequest(StreamInput in) throws IOException {
+ super(in);
+ local = in.readBoolean();
+ }
+
  @SuppressWarnings("unchecked")
  public final Request local(boolean local) {
  this.local = local;

server/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java

@@ -38,6 +38,11 @@ public abstract class MasterNodeRequest<Request extends MasterNodeRequest<Reques
  protected MasterNodeRequest() {
  }
 
+ protected MasterNodeRequest(StreamInput in) throws IOException {
+ super(in);
+ masterNodeTimeout = in.readTimeValue();
+ }
+
  /**
  * A timeout value in case the master has not been discovered yet or disconnected.
  */

server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.discovery.Discovery.FailedToCommitClusterStateException;
@@ -66,6 +67,12 @@ protected TransportMasterNodeAction(Settings settings, String actionName, Transp
  this(settings, actionName, true, transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver, request);
  }
 
+ protected TransportMasterNodeAction(Settings settings, String actionName, TransportService transportService,
+ ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
+ Writeable.Reader<Request> request, IndexNameExpressionResolver indexNameExpressionResolver) {
+ this(settings, actionName, true, transportService, clusterService, threadPool, actionFilters, request, indexNameExpressionResolver);
+ }
+
  protected TransportMasterNodeAction(Settings settings, String actionName, boolean canTripCircuitBreaker,
  TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
  ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
@@ -77,6 +84,17 @@ protected TransportMasterNodeAction(Settings settings, String actionName, boolea
  this.executor = executor();
  }
 
+ protected TransportMasterNodeAction(Settings settings, String actionName, boolean canTripCircuitBreaker,
+ TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
+ ActionFilters actionFilters, Writeable.Reader<Request> request,
+ IndexNameExpressionResolver indexNameExpressionResolver) {
+ super(settings, actionName, canTripCircuitBreaker, threadPool, transportService, actionFilters, request,
+ indexNameExpressionResolver);
+ this.transportService = transportService;
+ this.clusterService = clusterService;
+ this.executor = executor();
+ }
+
  protected abstract String executor();
 
  protected abstract Response newResponse();

server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeReadAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
@@ -49,6 +50,13 @@ protected TransportMasterNodeReadAction(Settings settings, String actionName, Tr
  this(settings, actionName, true, transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver,request);
  }
 
+ protected TransportMasterNodeReadAction(Settings settings, String actionName, TransportService transportService,
+ ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
+ Writeable.Reader<Request> request, IndexNameExpressionResolver indexNameExpressionResolver) {
+ this(settings, actionName, true, transportService, clusterService, threadPool, actionFilters, request,
+ indexNameExpressionResolver);
+ }
+
  protected TransportMasterNodeReadAction(Settings settings, String actionName, boolean checkSizeLimit, TransportService transportService,
  ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
  IndexNameExpressionResolver indexNameExpressionResolver, Supplier<Request> request) {
@@ -57,6 +65,14 @@ protected TransportMasterNodeReadAction(Settings settings, String actionName, bo
  this.forceLocal = FORCE_LOCAL_SETTING.get(settings);
  }
 
+ protected TransportMasterNodeReadAction(Settings settings, String actionName, boolean checkSizeLimit, TransportService transportService,
+ ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
+ Writeable.Reader<Request> request, IndexNameExpressionResolver indexNameExpressionResolver) {
+ super(settings, actionName, checkSizeLimit, transportService, clusterService, threadPool, actionFilters, request,
+ indexNameExpressionResolver);
+ this.forceLocal = FORCE_LOCAL_SETTING.get(settings);
+ }
+
  @Override
  protected final boolean localExecute(Request request) {
  return forceLocal || request.local();

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

@@ -55,8 +55,8 @@ public TransportCcrStatsAction(
  clusterService,
  threadPool,
  actionFilters,
- indexNameExpressionResolver,
- CcrStatsAction.Request::new
+ CcrStatsAction.Request::new,
+ indexNameExpressionResolver
  );
  this.client = client;
  this.ccrLicenseChecker = Objects.requireNonNull(ccrLicenseChecker);

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

@@ -37,7 +37,7 @@ public TransportDeleteAutoFollowPatternAction(Settings settings, TransportServic
  ThreadPool threadPool, ActionFilters actionFilters,
  IndexNameExpressionResolver indexNameExpressionResolver) {
  super(settings, DeleteAutoFollowPatternAction.NAME, transportService, clusterService, threadPool, actionFilters,
- indexNameExpressionResolver, DeleteAutoFollowPatternAction.Request::new);
+ DeleteAutoFollowPatternAction.Request::new, indexNameExpressionResolver);
  }
 
  @Override

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

@@ -38,7 +38,7 @@ public TransportGetAutoFollowPatternAction(Settings settings,
  ActionFilters actionFilters,
  IndexNameExpressionResolver indexNameExpressionResolver) {
  super(settings, GetAutoFollowPatternAction.NAME, transportService, clusterService, threadPool, actionFilters,
- indexNameExpressionResolver, GetAutoFollowPatternAction.Request::new);
+ GetAutoFollowPatternAction.Request::new, indexNameExpressionResolver);
  }
 
  @Override

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

@@ -40,7 +40,7 @@ public TransportPauseFollowAction(
  final ClusterService clusterService,
  final PersistentTasksService persistentTasksService) {
  super(settings, PauseFollowAction.NAME, transportService, clusterService, threadPool, actionFilters,
- indexNameExpressionResolver, PauseFollowAction.Request::new);
+ PauseFollowAction.Request::new, indexNameExpressionResolver);
  this.persistentTasksService = persistentTasksService;
  }
 

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

@@ -53,7 +53,7 @@ public TransportPutAutoFollowPatternAction(
  final IndexNameExpressionResolver indexNameExpressionResolver,
  final CcrLicenseChecker ccrLicenseChecker) {
  super(settings, PutAutoFollowPatternAction.NAME, transportService, clusterService, threadPool, actionFilters,
- indexNameExpressionResolver, PutAutoFollowPatternAction.Request::new);
+ PutAutoFollowPatternAction.Request::new, indexNameExpressionResolver);
  this.client = client;
  this.ccrLicenseChecker = Objects.requireNonNull(ccrLicenseChecker, "ccrLicenseChecker");
  }