PR Fixes - Enforce local only use for status action

Author: lukewhiting

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/StreamsStatusAction.java

@@ -11,8 +11,8 @@
 
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.action.support.local.LocalClusterStateRequest;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.tasks.CancellableTask;
@@ -47,13 +47,9 @@ public Response(boolean logsEnabled) {
             logs_enabled = logsEnabled;
         }
 
-        public Response(StreamInput in) throws IOException {
-            logs_enabled = in.readBoolean();
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeBoolean(logs_enabled);
+            TransportAction.localOnly();
         }
 
         @Override

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/TransportLogsStreamsToggleActivation.java

@@ -18,7 +18,7 @@
 import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.SequentialTaskAckingTaskExecutor;
+import org.elasticsearch.cluster.SequentialAckingBatchedTaskExecutor;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.ProjectId;
@@ -64,7 +64,7 @@ public TransportLogsStreamsToggleActivation(
         this.taskQueue = clusterService.createTaskQueue(
             "streams-update-state-queue",
             Priority.NORMAL,
-            new SequentialTaskAckingTaskExecutor<>()
+            new SequentialAckingBatchedTaskExecutor<>()
         );
         this.projectResolver = projectResolver;
     }

…er/SequentialTaskAckingTaskExecutor.java …SequentialAckingBatchedTaskExecutor.javaserver/src/main/java/org/elasticsearch/cluster/SequentialTaskAckingTaskExecutor.java renamed to server/src/main/java/org/elasticsearch/cluster/SequentialAckingBatchedTaskExecutor.java server/src/main/java/org/elasticsearch/cluster/SequentialTaskAckingTaskExecutor.java renamed to server/src/main/java/org/elasticsearch/cluster/SequentialAckingBatchedTaskExecutor.java

@@ -18,7 +18,8 @@
  *
  * @param <Task> The type of the task that extends {@link AckedClusterStateUpdateTask}.
  */
-public class SequentialTaskAckingTaskExecutor<Task extends AckedClusterStateUpdateTask> extends SimpleBatchedAckListenerTaskExecutor<Task> {
+public class SequentialAckingBatchedTaskExecutor<Task extends AckedClusterStateUpdateTask> extends SimpleBatchedAckListenerTaskExecutor<
+    Task> {
     @Override
     public Tuple<ClusterState, ClusterStateAckListener> executeTask(Task task, ClusterState clusterState) throws Exception {
         return Tuple.tuple(task.execute(clusterState), task);