Add null check in more places for persistent tasks (#123566)

There is no guarantee that a project has non-null persistent tasks [0].
Null check is already done in most places. This PR adds it in a few more
places.

[0] Since health-node is now a cluster-scoped persistent task, it has
become more likely for the project-scoped tasks to be null.

Relates: #123262

Author: ywangd

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/GetMigrationReindexStatusTransportAction.java

@@ -66,11 +66,10 @@ public GetMigrationReindexStatusTransportAction(
     protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
         String index = request.getIndex();
         String persistentTaskId = ReindexDataStreamAction.TASK_ID_PREFIX + index;
-        PersistentTasksCustomMetadata persistentTasksCustomMetadata = clusterService.state()
-            .getMetadata()
-            .getProject()
-            .custom(PersistentTasksCustomMetadata.TYPE);
-        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = persistentTasksCustomMetadata.getTask(persistentTaskId);
+        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
+            clusterService.state(),
+            persistentTaskId
+        );
         if (persistentTask == null) {
             listener.onFailure(new ResourceNotFoundException("No migration reindex status found for [{}]", index));
         } else if (persistentTask.isAssigned()) {

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java

@@ -297,11 +297,8 @@ private TimeValue updateCompletionTimeAndGetTimeToLive(
         ReindexDataStreamTask reindexDataStreamTask,
         @Nullable ReindexDataStreamPersistentTaskState state
     ) {
-        PersistentTasksCustomMetadata persistentTasksCustomMetadata = clusterService.state()
-            .getMetadata()
-            .getProject()
-            .custom(PersistentTasksCustomMetadata.TYPE);
-        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = persistentTasksCustomMetadata.getTask(
+        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
+            clusterService.state(),
             reindexDataStreamTask.getPersistentTaskId()
         );
         if (persistentTask == null) {

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamTask.java

@@ -66,13 +66,12 @@ public ReindexDataStreamTask(
 
     @Override
     public ReindexDataStreamStatus getStatus() {
-        PersistentTasksCustomMetadata persistentTasksCustomMetadata = clusterService.state()
-            .getMetadata()
-            .getProject()
-            .custom(PersistentTasksCustomMetadata.TYPE);
         int totalIndices = initialTotalIndices;
         int totalIndicesToBeUpgraded = initialTotalIndicesToBeUpgraded;
-        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = persistentTasksCustomMetadata.getTask(getPersistentTaskId());
+        PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
+            clusterService.state(),
+            getPersistentTaskId()
+        );
         if (persistentTask != null) {
             ReindexDataStreamPersistentTaskState state = (ReindexDataStreamPersistentTaskState) persistentTask.getState();
             if (state != null && state.totalIndices() != null && state.totalIndicesToBeUpgraded() != null) {

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportSetTransformUpgradeModeAction.java

@@ -110,6 +110,10 @@ private void stopTransforms(SetUpgradeModeActionRequest request, ClusterState st
 
     private void unassignTransforms(ClusterState state, ActionListener<Void> listener) {
         PersistentTasksCustomMetadata tasksCustomMetadata = state.metadata().getProject().custom(PersistentTasksCustomMetadata.TYPE);
+        if (tasksCustomMetadata == null) {
+            listener.onResponse(null);
+            return;
+        }
         var transformTasks = tasksCustomMetadata.tasks()
             .stream()
             .filter(this::isTransformTask)