Review fixes

Author: GalLalouche

server/src/main/java/org/elasticsearch/node/Node.java

@@ -357,7 +357,6 @@ public Node start() throws NodeValidationException {
         clusterService.start();
         assert clusterService.localNode().equals(localNodeFactory.getNode())
             : "clusterService has a different local node than the factory provided";
-        transportService.getTaskManager().setLocalNodeId(clusterService.localNode().getId());
         transportService.acceptIncomingRequests();
         /*
          * CoordinationDiagnosticsService expects to be able to send transport requests and use the cluster state, so it is important to

server/src/main/java/org/elasticsearch/tasks/TaskAwareRequest.java

@@ -9,6 +9,8 @@
 
 package org.elasticsearch.tasks;
 
+import org.elasticsearch.core.Nullable;
+
 import java.util.Map;
 
 /**
@@ -56,7 +58,14 @@ default Task createTask(long id, String type, String action, TaskId parentTaskId
      * Returns the task object that should be used to keep track of the processing of the request, with an extra local node ID.
      */
     // TODO remove the above overload, use only this one.
-    default Task createTask(String localNodeId, long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+    default Task createTask(
+        @Nullable String localNodeId,
+        long id,
+        String type,
+        String action,
+        TaskId parentTaskId,
+        Map<String, String> headers
+    ) {
         return createTask(id, type, action, parentTaskId, headers);
     }
 

server/src/main/java/org/elasticsearch/tasks/TaskManager.java

@@ -87,7 +87,6 @@ public class TaskManager implements ClusterStateApplier {
     private final ByteSizeValue maxHeaderSize;
     private final Map<TcpChannel, ChannelPendingTaskTracker> channelPendingTaskTrackers = ConcurrentCollections.newConcurrentMap();
     private final SetOnce<TaskCancellationService> cancellationService = new SetOnce<>();
-    private final SetOnce<String> localNodeId = new SetOnce<>();
 
     private final List<RemovedTaskListener> removedTaskListeners = new CopyOnWriteArrayList<>();
 
@@ -112,10 +111,6 @@ public void setTaskCancellationService(TaskCancellationService taskCancellationS
         this.cancellationService.set(taskCancellationService);
     }
 
-    public void setLocalNodeId(String localNodeId) {
-        this.localNodeId.set(localNodeId);
-    }
-
     /**
      * Registers a task without parent task
      */
@@ -146,10 +141,14 @@ public Task register(String type, String action, TaskAwareRequest request, boole
                 headers.put(key, httpHeader);
             }
         }
-        var localNodeId = this.localNodeId.get();
-        Task task = localNodeId != null
-            ? request.createTask(localNodeId, taskIdGenerator.incrementAndGet(), type, action, request.getParentTask(), headers)
-            : request.createTask(taskIdGenerator.incrementAndGet(), type, action, request.getParentTask(), headers);
+        Task task = request.createTask(
+            lastDiscoveryNodes.getLocalNodeId(),
+            taskIdGenerator.incrementAndGet(),
+            type,
+            action,
+            request.getParentTask(),
+            headers
+        );
         Objects.requireNonNull(task);
         assert task.getParentTaskId().equals(request.getParentTask()) : "Request [ " + request + "] didn't preserve it parentTaskId";
         if (logger.isTraceEnabled()) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlQueryStatus.java

@@ -34,7 +34,7 @@ private EsqlQueryStatus(StreamInput stream) throws IOException {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        out.writeWriteable(id);
+        id.writeTo(out);
     }
 
     @Override