Use TaskId in TaskAwareRequest.createRequest

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

@@ -56,8 +56,8 @@ public boolean shouldEnable() {
         }
 
         @Override
-        public Task createTask(String localNodeId, long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
-            return new CancellableTask(id, type, action, "Logs streams activation toggle request", parentTaskId, headers);
+        public Task createTask(TaskId taskId, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(taskId.getId(), type, action, "Logs streams activation toggle request", parentTaskId, headers);
         }
     }
 }

plugins/discovery-ec2/src/test/java/org/elasticsearch/discovery/ec2/EC2RetriesTests.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.discovery.SeedHostsResolver;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.test.transport.MockTransportService;
-import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.netty4.Netty4Transport;
 import org.elasticsearch.transport.netty4.SharedGroupFactory;
 import org.hamcrest.Matchers;
@@ -47,8 +46,7 @@ public class EC2RetriesTests extends AbstractEC2MockAPITestCase {
 
     @Override
     protected MockTransportService createTransportService() {
-        return new MockTransportService(
-            Settings.EMPTY,
+        return MockTransportService.createMockTransportService(
             new Netty4Transport(
                 Settings.EMPTY,
                 TransportVersion.current(),
@@ -59,9 +57,7 @@ protected MockTransportService createTransportService() {
                 new NoneCircuitBreakerService(),
                 new SharedGroupFactory(Settings.EMPTY)
             ),
-            threadPool,
-            TransportService.NOOP_TRANSPORT_INTERCEPTOR,
-            null
+            threadPool
         );
     }
 

plugins/discovery-ec2/src/test/java/org/elasticsearch/discovery/ec2/Ec2DiscoveryTests.java

@@ -78,7 +78,7 @@ public TransportAddress[] addressesFromString(String address) {
                 return new TransportAddress[] { poorMansDNS.getOrDefault(address, buildNewFakeTransportAddress()) };
             }
         };
-        return new MockTransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, null);
+        return MockTransportService.createMockTransportService(transport, threadPool);
     }
 
     protected List<TransportAddress> buildDynamicHosts(Settings nodeSettings, int nodes) {

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

@@ -685,14 +685,16 @@ private void construct(
 
         modules.bindToInstance(Tracer.class, telemetryProvider.getTracer());
 
+        assert nodeEnvironment.nodeId() != null : "node ID must be set before constructing the Node";
         TaskManager taskManager = new TaskManager(
             settings,
             threadPool,
             Stream.concat(
                 pluginsService.filterPlugins(ActionPlugin.class).flatMap(p -> p.getTaskHeaders().stream()),
                 Task.HEADERS_TO_COPY.stream()
             ).collect(Collectors.toSet()),
-            telemetryProvider.getTracer()
+            telemetryProvider.getTracer(),
+            nodeEnvironment.nodeId()
         );
 
         ClusterService clusterService = createClusterService(settingsModule, threadPool, taskManager);
@@ -1097,7 +1099,8 @@ public Map<String, String> queryFields() {
             localNodeFactory,
             settingsModule.getClusterSettings(),
             taskManager,
-            telemetryProvider.getTracer()
+            telemetryProvider.getTracer(),
+            nodeEnvironment.nodeId()
         );
         final SearchResponseMetrics searchResponseMetrics = new SearchResponseMetrics(telemetryProvider.getMeterRegistry());
         final SearchTransportService searchTransportService = new SearchTransportService(

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

@@ -114,7 +114,8 @@ TransportService newTransportService(
         Function<BoundTransportAddress, DiscoveryNode> localNodeFactory,
         ClusterSettings clusterSettings,
         TaskManager taskManager,
-        Tracer tracer
+        Tracer tracer,
+        String nodeId
     ) {
         return new TransportService(settings, transport, threadPool, interceptor, localNodeFactory, clusterSettings, taskManager, tracer);
     }

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

@@ -9,8 +9,6 @@
 
 package org.elasticsearch.tasks;
 
-import org.elasticsearch.core.Nullable;
-
 import java.util.Map;
 
 /**
@@ -57,18 +55,8 @@ 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(
-        // TODO this is only nullable in tests, where the MockNode does not guarantee the localNodeId is set before calling this method. We
-        // We should fix the tests, and replace this and id with TaskId instead.
-        @Nullable String localNodeId,
-        long id,
-        String type,
-        String action,
-        TaskId parentTaskId,
-        Map<String, String> headers
-    ) {
-        return createTask(id, type, action, parentTaskId, headers);
+    default Task createTask(TaskId taskId, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return createTask(taskId.getId(), type, action, parentTaskId, headers);
     }
 
     /**

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

@@ -21,6 +21,7 @@
 import org.elasticsearch.cluster.ClusterStateApplier;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
@@ -80,6 +81,7 @@ public class TaskManager implements ClusterStateApplier {
 
     private TaskResultsService taskResultsService;
 
+    private final String nodeId;
     private DiscoveryNodes lastDiscoveryNodes = DiscoveryNodes.EMPTY_NODES;
 
     private final Tracer tracer;
@@ -95,11 +97,19 @@ public TaskManager(Settings settings, ThreadPool threadPool, Set<String> taskHea
         this(settings, threadPool, taskHeaders, Tracer.NOOP);
     }
 
+    // For testing (especially the creating a random node ID, which some tests rely on)
     public TaskManager(Settings settings, ThreadPool threadPool, Set<String> taskHeaders, Tracer tracer) {
+        this(settings, threadPool, taskHeaders, tracer, UUIDs.randomBase64UUID());
+    }
+
+    // TODO Both of the above overloads should be moved to the test package.
+
+    public TaskManager(Settings settings, ThreadPool threadPool, Set<String> taskHeaders, Tracer tracer, String nodeId) {
         this.threadPool = threadPool;
         this.taskHeaders = Set.copyOf(taskHeaders);
         this.maxHeaderSize = SETTING_HTTP_MAX_HEADER_SIZE.get(settings);
         this.tracer = tracer;
+        this.nodeId = nodeId;
     }
 
     public void setTaskResultsService(TaskResultsService taskResultsService) {
@@ -142,8 +152,7 @@ public Task register(String type, String action, TaskAwareRequest request, boole
             }
         }
         Task task = request.createTask(
-            lastDiscoveryNodes.getLocalNodeId(),
-            taskIdGenerator.incrementAndGet(),
+            new TaskId(nodeId, taskIdGenerator.incrementAndGet()),
             type,
             action,
             request.getParentTask(),

server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TaskManagerTestCase.java

@@ -177,7 +177,7 @@ public TestNode(String name, ThreadPool threadPool, Settings settings) {
                 discoveryNode.set(DiscoveryNodeUtils.create(name, address.publishAddress(), emptyMap(), emptySet()));
                 return discoveryNode.get();
             };
-            TaskManager taskManager = createTaskManager(settings, threadPool, emptySet(), Tracer.NOOP);
+            TaskManager taskManager = createTaskManager(settings, threadPool, emptySet(), Tracer.NOOP, name);
             transportService = new TransportService(
                 settings,
                 new Netty4Transport(

server/src/test/java/org/elasticsearch/action/resync/TransportResyncReplicationActionTests.java

@@ -123,7 +123,8 @@ public void testResyncDoesNotBlockOnPrimaryAction() throws Exception {
                     NOOP_TRANSPORT_INTERCEPTOR,
                     x -> clusterService.localNode(),
                     null,
-                    Collections.emptySet()
+                    Collections.emptySet(),
+                    clusterService.localNode().getId()
                 );
                 transportService.start();
                 transportService.acceptIncomingRequests();

server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationActionTests.java

@@ -1316,7 +1316,9 @@ public void testRetryOnReplicaWithRealTransport() throws Exception {
             TransportService.NOOP_TRANSPORT_INTERCEPTOR,
             x -> clusterService.localNode(),
             null,
-            Collections.emptySet()
+            Collections.emptySet(),
+            clusterService.localNode().getId()
+
         );
         transportService.start();
         transportService.acceptIncomingRequests();

server/src/test/java/org/elasticsearch/discovery/FileBasedSeedHostsProviderTests.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TcpTransport;
-import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.netty4.Netty4Transport;
 import org.elasticsearch.transport.netty4.SharedGroupFactory;
 import org.junit.After;
@@ -87,13 +86,7 @@ public BoundTransportAddress boundAddress() {
                 );
             }
         };
-        transportService = new MockTransportService(
-            Settings.EMPTY,
-            transport,
-            threadPool,
-            TransportService.NOOP_TRANSPORT_INTERCEPTOR,
-            null
-        );
+        transportService = MockTransportService.createMockTransportService(transport, threadPool);
     }
 
     public void testBuildDynamicNodes() throws Exception {

server/src/test/java/org/elasticsearch/transport/TransportServiceHandshakeTests.java

@@ -92,7 +92,8 @@ private TransportService startServices(
                 .version(nodeVersion)
                 .build(),
             null,
-            Collections.emptySet()
+            Collections.emptySet(),
+            nodeNameAndId
         );
         transportService.start();
         transportService.acceptIncomingRequests();

test/framework/src/main/java/org/elasticsearch/node/MockNode.java

@@ -167,8 +167,10 @@ protected TransportService newTransportService(
             Function<BoundTransportAddress, DiscoveryNode> localNodeFactory,
             ClusterSettings clusterSettings,
             TaskManager taskManager,
-            Tracer tracer
+            Tracer tracer,
+            String nodeId
         ) {
+
             // we use the MockTransportService.TestPlugin class as a marker to create a network
             // module with this MockNetworkService. NetworkService is such an integral part of the systme
             // we don't allow to plug it in from plugins or anything. this is a test-only override and
@@ -183,7 +185,8 @@ protected TransportService newTransportService(
                     localNodeFactory,
                     clusterSettings,
                     taskManager,
-                    tracer
+                    tracer,
+                    nodeId
                 );
             } else {
                 return new MockTransportService(
@@ -193,7 +196,8 @@ protected TransportService newTransportService(
                     interceptor,
                     localNodeFactory,
                     clusterSettings,
-                    taskManager.getTaskHeaders()
+                    taskManager.getTaskHeaders(),
+                    nodeId
                 );
             }
         }

test/framework/src/main/java/org/elasticsearch/test/ClusterServiceUtils.java

@@ -160,7 +160,7 @@ public static ClusterService createClusterService(
             settings,
             clusterSettings,
             threadPool,
-            new TaskManager(settings, threadPool, Collections.emptySet(), Tracer.NOOP)
+            new TaskManager(settings, threadPool, Collections.emptySet(), Tracer.NOOP, localNode.getId())
         );
         clusterService.setNodeConnectionsService(createNoOpNodeConnectionsService());
         ClusterState.Builder builder = ClusterState.builder(new ClusterName(ClusterServiceUtils.class.getSimpleName()))