[ML] Fix NPE in ML assignment notifier (#107312)

Fix NPE in ML assignment notifier checking for unassigned tasks.
The error is harmless as when the persistent task acquires state
the check will complete normally.

docs/changelog/107312.yaml

@@ -0,0 +1,5 @@
+pr: 107312
+summary: Fix NPE in ML assignment notifier
+area: Machine Learning
+type: bug
+issues: []

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

@@ -300,7 +300,7 @@ public class MlAssignmentNotifier implements ClusterStateListener {
                 final String taskName = task.getTaskName();
                 if (MlTasks.JOB_TASK_NAME.equals(taskName) || MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME.equals(taskName)) {
                     // Ignore failed tasks - they don't need to be assigned to a node
-                    if (((MlTaskState) task.getState()).isFailed()) {
+                    if (task.getState() == null || ((MlTaskState) task.getState()).isFailed()) {
                         continue;
                     }
                     final String mlId = ((MlTaskParams) task.getParams()).getMlId();

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlAssignmentNotifierTests.java

@@ -317,4 +317,32 @@ public class MlAssignmentNotifierTests extends ESTestCase {
         // as it doesn't make any difference whether they're assigned or not and autoscaling will ignore them
         assertThat(itemsToReport, empty());
     }
+
+    public void testFindLongTimeUnassignedTasks_WithNullState() {
+        MlAssignmentNotifier notifier = new MlAssignmentNotifier(
+            anomalyDetectionAuditor,
+            dataFrameAnalyticsAuditor,
+            threadPool,
+            clusterService
+        );
+        var now = Instant.now();
+        var sevenHoursAgo = now.minus(Duration.ofHours(7));
+        var eightHoursAgo = now.minus(Duration.ofHours(8));
+
+        {
+            // run once with valid state to add unassigned job to the history
+            PersistentTasksCustomMetadata.Builder tasksBuilder = PersistentTasksCustomMetadata.builder();
+            addJobTask("job1", null, JobState.OPENED, tasksBuilder);
+            List<String> itemsToReport = notifier.findLongTimeUnassignedTasks(eightHoursAgo, tasksBuilder.build());
+            // Nothing reported because unassigned jobs only just detected
+            assertThat(itemsToReport, empty());
+        }
+        {
+            PersistentTasksCustomMetadata.Builder tasksBuilder = PersistentTasksCustomMetadata.builder();
+            addJobTask("job1", null, null, tasksBuilder); // this time the job has no state
+            // one hour later the job would be detected as unassigned if not for the missing state
+            List<String> itemsToReport = notifier.findLongTimeUnassignedTasks(sevenHoursAgo, tasksBuilder.build());
+            assertThat(itemsToReport, empty());
+        }
+    }
 }