[ML] fix ml API cancellation when http connection is closed (#88616)

API cancellation was erroneously using the node-name for the parent task
id. Task ids are constructed via the node ID, not name. 

This fixes that bug.

related to:   - #88030  -
#88142  -
#88009

Author: benwtrent

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportEvaluateDataFrameAction.java

@@ -74,7 +74,7 @@ protected void doExecute(
         EvaluateDataFrameAction.Request request,
         ActionListener<EvaluateDataFrameAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         ActionListener<List<Void>> resultsListener = ActionListener.wrap(unused -> {
             EvaluateDataFrameAction.Response response = new EvaluateDataFrameAction.Response(
                 request.getEvaluation().getName(),

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportExplainDataFrameAnalyticsAction.java

@@ -125,7 +125,7 @@ private void explain(
         ExplainDataFrameAnalyticsAction.Request request,
         ActionListener<ExplainDataFrameAnalyticsAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         final ExtractedFieldsDetectorFactory extractedFieldsDetectorFactory = new ExtractedFieldsDetectorFactory(
             new ParentTaskAssigningClient(client, parentTaskId)
         );

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java

@@ -46,7 +46,7 @@ public TransportGetCategoriesAction(
 
     @Override
     protected void doExecute(Task task, GetCategoriesAction.Request request, ActionListener<GetCategoriesAction.Response> listener) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         jobManager.jobExists(request.getJobId(), parentTaskId, ActionListener.wrap(jobExists -> {
             Integer from = request.getPageParams() != null ? request.getPageParams().getFrom() : null;
             Integer size = request.getPageParams() != null ? request.getPageParams().getSize() : null;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetDataFrameAnalyticsStatsAction.java

@@ -158,7 +158,7 @@ protected void doExecute(
         GetDataFrameAnalyticsStatsAction.Request request,
         ActionListener<GetDataFrameAnalyticsStatsAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         logger.debug("Get stats for data frame analytics [{}]", request.getId());
 
         ActionListener<GetDataFrameAnalyticsAction.Response> getResponseListener = ActionListener.wrap(getResponse -> {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetDatafeedsAction.java

@@ -61,7 +61,7 @@ protected void masterOperation(
         ClusterState state,
         ActionListener<GetDatafeedsAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         logger.debug("Get datafeed '{}'", request.getDatafeedId());
 
         datafeedManager.getDatafeeds(

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetDatafeedsStatsAction.java

@@ -68,7 +68,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
         ClusterState state = clusterService.state();
         final PersistentTasksCustomMetadata tasksInProgress = state.getMetadata().custom(PersistentTasksCustomMetadata.TYPE);
         final Response.Builder responseBuilder = new Response.Builder();
-        final TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        final TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
 
         // 5. Build response
         ActionListener<GetDatafeedRunningStateAction.Response> runtimeStateListener = ActionListener.wrap(runtimeStateResponse -> {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobModelSnapshotsUpgradeStatsAction.java

@@ -76,7 +76,7 @@ protected void masterOperation(Task task, Request request, ClusterState state, A
         logger.debug(() -> format("[%s] get stats for model snapshot [%s] upgrades", request.getJobId(), request.getSnapshotId()));
         final PersistentTasksCustomMetadata tasksInProgress = state.getMetadata().custom(PersistentTasksCustomMetadata.TYPE);
         final Collection<PersistentTasksCustomMetadata.PersistentTask<?>> snapshotUpgrades = MlTasks.snapshotUpgradeTasks(tasksInProgress);
-        final TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        final TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
 
         // 2. Now that we have the job IDs, find the relevant model snapshot upgrades
         ActionListener<List<Job.Builder>> expandIdsListener = ActionListener.wrap(jobs -> {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsAction.java

@@ -69,7 +69,7 @@ protected void masterOperation(
         ClusterState state,
         ActionListener<GetJobsAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         logger.debug("Get job '{}'", request.getJobId());
         jobManager.expandJobBuilders(
             request.getJobId(),

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetJobsStatsAction.java

@@ -95,7 +95,7 @@ public TransportGetJobsStatsAction(
     @Override
     protected void doExecute(Task task, GetJobsStatsAction.Request request, ActionListener<GetJobsStatsAction.Response> finalListener) {
         logger.debug("Get stats for job [{}]", request.getJobId());
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
 
         ClusterState state = clusterService.state();
         PersistentTasksCustomMetadata tasks = state.getMetadata().custom(PersistentTasksCustomMetadata.TYPE);
@@ -144,7 +144,7 @@ protected void taskOperation(
         JobTask task,
         ActionListener<QueryPage<JobStats>> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), actionTask.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), actionTask.getId());
         String jobId = task.getJobId();
         ClusterState state = clusterService.state();
         PersistentTasksCustomMetadata tasks = state.getMetadata().custom(PersistentTasksCustomMetadata.TYPE);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java

@@ -53,7 +53,7 @@ protected void doExecute(
         GetModelSnapshotsAction.Request request,
         ActionListener<GetModelSnapshotsAction.Response> listener
     ) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         logger.debug(
             () -> format(
                 "Get model snapshots for job %s snapshot ID %s. from = %s, size = %s start = '%s', end='%s', sort=%s descending=%s",

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportInferTrainedModelDeploymentAction.java

@@ -72,7 +72,7 @@ protected void doExecute(
         InferTrainedModelDeploymentAction.Request request,
         ActionListener<InferTrainedModelDeploymentAction.Response> listener
     ) {
-        TaskId taskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId taskId = new TaskId(clusterService.localNode().getId(), task.getId());
         final String deploymentId = request.getDeploymentId();
         // We need to check whether there is at least an assigned task here, otherwise we cannot redirect to the
         // node running the job task.

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportInternalInferModelAction.java

@@ -96,7 +96,7 @@ public TransportInternalInferModelAction(
     protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
 
         Response.Builder responseBuilder = Response.builder();
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
 
         if (MachineLearningField.ML_API_FEATURE.check(licenseState)) {
             responseBuilder.setLicensed(true);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPreviewDataFrameAnalyticsAction.java

@@ -102,7 +102,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
     }
 
     void preview(Task task, DataFrameAnalyticsConfig config, ActionListener<Response> listener) {
-        final TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        final TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         final ExtractedFieldsDetectorFactory extractedFieldsDetectorFactory = new ExtractedFieldsDetectorFactory(
             new ParentTaskAssigningClient(client, parentTaskId)
         );

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPreviewDatafeedAction.java

@@ -92,7 +92,7 @@ public TransportPreviewDatafeedAction(
 
     @Override
     protected void doExecute(Task task, PreviewDatafeedAction.Request request, ActionListener<PreviewDatafeedAction.Response> listener) {
-        TaskId parentTaskId = new TaskId(clusterService.getNodeName(), task.getId());
+        TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
         ActionListener<DatafeedConfig> datafeedConfigActionListener = ActionListener.wrap(datafeedConfig -> {
             if (request.getJobConfig() != null) {
                 previewDatafeed(parentTaskId, datafeedConfig, request.getJobConfig().build(new Date()), request, listener);