[CCR] Change resume follow api to be a master node action (elastic#35249)

In order to start shard follow tasks, the resume follow api already needs execute N requests to the elected master node. The pause follow API is also a master node action, which would make how both APIs execute more consistent.

Author: martijnvg

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

@@ -8,10 +8,12 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
@@ -50,7 +52,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-public class TransportResumeFollowAction extends HandledTransportAction<ResumeFollowAction.Request, AcknowledgedResponse> {
+public class TransportResumeFollowAction extends TransportMasterNodeAction<ResumeFollowAction.Request, AcknowledgedResponse> {
 
  static final ByteSizeValue DEFAULT_MAX_READ_REQUEST_SIZE = new ByteSizeValue(32, ByteSizeUnit.MB);
  static final ByteSizeValue DEFAULT_MAX_WRITE_REQUEST_SIZE = new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES);
@@ -64,7 +66,6 @@ public class TransportResumeFollowAction extends HandledTransportAction<ResumeFo
  static final TimeValue DEFAULT_READ_POLL_TIMEOUT = TimeValue.timeValueMinutes(1);
 
  private final Client client;
- private final ClusterService clusterService;
  private final PersistentTasksService persistentTasksService;
  private final IndicesService indicesService;
  private final CcrLicenseChecker ccrLicenseChecker;
@@ -81,24 +82,38 @@ public TransportResumeFollowAction(
  final PersistentTasksService persistentTasksService,
  final IndicesService indicesService,
  final CcrLicenseChecker ccrLicenseChecker) {
- super(settings, ResumeFollowAction.NAME, threadPool, transportService, actionFilters,
+ super(settings, ResumeFollowAction.NAME, true, transportService, clusterService, threadPool, actionFilters,
  ResumeFollowAction.Request::new, indexNameExpressionResolver);
  this.client = client;
- this.clusterService = clusterService;
  this.persistentTasksService = persistentTasksService;
  this.indicesService = indicesService;
  this.ccrLicenseChecker = Objects.requireNonNull(ccrLicenseChecker);
  }
 
  @Override
- protected void doExecute(final ResumeFollowAction.Request request,
- final ActionListener<AcknowledgedResponse> listener) {
+ protected String executor() {
+ return ThreadPool.Names.SAME;
+ }
+
+ @Override
+ protected AcknowledgedResponse newResponse() {
+ return new AcknowledgedResponse();
+ }
+
+ @Override
+ protected ClusterBlockException checkBlock(ResumeFollowAction.Request request, ClusterState state) {
+ return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+ }
+
+ @Override
+ protected void masterOperation(final ResumeFollowAction.Request request,
+ ClusterState state,
+ final ActionListener<AcknowledgedResponse> listener) throws Exception {
  if (ccrLicenseChecker.isCcrAllowed() == false) {
  listener.onFailure(LicenseUtils.newComplianceException("ccr"));
  return;
  }
 
- final ClusterState state = clusterService.state();
  final IndexMetaData followerIndexMetadata = state.getMetaData().index(request.getFollowerIndex());
  if (followerIndexMetadata == null) {
  listener.onFailure(new IndexNotFoundException(request.getFollowerIndex()));

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

@@ -7,11 +7,11 @@
 package org.elasticsearch.xpack.core.ccr.action;
 
 import org.elasticsearch.action.Action;
-import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.ElasticsearchClient;
+import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -46,7 +46,7 @@ public AcknowledgedResponse newResponse() {
  return new AcknowledgedResponse();
  }
 
- public static class Request extends ActionRequest implements ToXContentObject {
+ public static class Request extends MasterNodeRequest<Request> implements ToXContentObject {
 
  static final ParseField FOLLOWER_INDEX_FIELD = new ParseField("follower_index");
  static final ParseField MAX_READ_REQUEST_OPERATION_COUNT = new ParseField("max_read_request_operation_count");