Enable Setting Master Node Timeout in Watcher Start/Stop Requests (#70425)

It's in the title, we have to be able to set this timeout. Otherwise,
it's impossible to deactive/active watcher or a slow master node.
In the worst case scenario, Wacher may be at fault for making the master slow
and it becomes impossible to deactive it.

x-pack/docs/en/rest-api/watcher/start.asciidoc

@@ -24,8 +24,13 @@ information, see <<security-privileges>>.
 //[[watcher-api-start-path-params]]
 //==== {api-path-parms-title}
 
-//[[watcher-api-start-query-params]]
-//==== {api-query-parms-title}
+[[watcher-api-start-query-params]]
+==== {api-query-parms-title}
+
+`master_timeout`::
+(Optional, <<time-units, time units>>) Specifies the period of time to wait for
+a connection to the master node. If no response is received before the timeout
+expires, the request fails and returns an error. Defaults to `30s`.
 
 //[[watcher-api-start-request-body]]
 //==== {api-request-body-title}

x-pack/docs/en/rest-api/watcher/stop.asciidoc

@@ -24,8 +24,13 @@ information, see <<security-privileges>>.
 //[[watcher-api-stop-path-params]]
 //==== {api-path-parms-title}
 
-//[[watcher-api-stop-query-params]]
-//==== {api-query-parms-title}
+[[watcher-api-stop-query-params]]
+==== {api-query-parms-title}
+
+`master_timeout`::
+(Optional, <<time-units, time units>>) Specifies the period of time to wait for
+a connection to the master node. If no response is received before the timeout
+expires, the request fails and returns an error. Defaults to `30s`.
 
 //[[watcher-api-stop-request-body]]
 //==== {api-request-body-title}

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/rest/action/RestWatchServiceAction.java

@@ -49,9 +49,10 @@ public class RestWatchServiceAction extends BaseRestHandler {
         }
 
         @Override
-        public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
-            return channel ->
-                client.execute(WatcherServiceAction.INSTANCE, new WatcherServiceRequest().stop(), new RestToXContentListener<>(channel));
+        public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) {
+            final WatcherServiceRequest request = new WatcherServiceRequest().stop();
+            request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
+            return channel -> client.execute(WatcherServiceAction.INSTANCE, request, new RestToXContentListener<>(channel));
         }
     }
 }

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/TransportWatcherServiceAction.java

@@ -36,18 +36,6 @@ public class TransportWatcherServiceAction extends AcknowledgedTransportMasterNo
 
     private static final Logger logger = LogManager.getLogger(TransportWatcherServiceAction.class);
 
-    private static final AckedRequest ackedRequest = new AckedRequest() {
-        @Override
-        public TimeValue ackTimeout() {
-            return AcknowledgedRequest.DEFAULT_ACK_TIMEOUT;
-        }
-
-        @Override
-        public TimeValue masterNodeTimeout() {
-            return AcknowledgedRequest.DEFAULT_ACK_TIMEOUT;
-        }
-    };
-
     @Inject
     public TransportWatcherServiceAction(TransportService transportService, ClusterService clusterService,
                                          ThreadPool threadPool, ActionFilters actionFilters,
@@ -59,20 +47,22 @@ public class TransportWatcherServiceAction extends AcknowledgedTransportMasterNo
     @Override
     protected void masterOperation(Task task, WatcherServiceRequest request, ClusterState state,
                                    ActionListener<AcknowledgedResponse> listener) {
-        switch (request.getCommand()) {
-            case STOP:
-                setWatcherMetadataAndWait(true, listener);
-                break;
-            case START:
-                setWatcherMetadataAndWait(false, listener);
-                break;
-        }
-    }
+        final boolean manuallyStopped = request.getCommand() == WatcherServiceRequest.Command.STOP;
+        final String source = manuallyStopped ? "update_watcher_manually_stopped" : "update_watcher_manually_started";
 
-    private void setWatcherMetadataAndWait(boolean manuallyStopped, final ActionListener<AcknowledgedResponse> listener) {
-        String source = manuallyStopped ? "update_watcher_manually_stopped" : "update_watcher_manually_started";
+        // TODO: make WatcherServiceRequest a real AckedRequest so that we have both a configurable timeout and master node timeout like
+        //       we do elsewhere
+        clusterService.submitStateUpdateTask(source, new AckedClusterStateUpdateTask(new AckedRequest() {
+            @Override
+            public TimeValue ackTimeout() {
+                return AcknowledgedRequest.DEFAULT_ACK_TIMEOUT;
+            }
 
-        clusterService.submitStateUpdateTask(source, new AckedClusterStateUpdateTask(ackedRequest, listener) {
+            @Override
+            public TimeValue masterNodeTimeout() {
+                return request.masterNodeTimeout();
+            }
+        }, listener) {
             @Override
             public ClusterState execute(ClusterState clusterState) {
                 XPackPlugin.checkReadyForXPackCustomMetadata(clusterState);