Protect against project deletion

Author: nielsbauman

server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java

@@ -188,7 +188,8 @@ public void waitForPersistentTaskCondition(
      *
      * @param projectId the project ID
      * @param taskId the persistent task id
-     * @param predicate the persistent task predicate to evaluate
+     * @param predicate the persistent task predicate to evaluate, must be able to handle {@code null} input which means either the project
+     *                  does not exist or persistent tasks for the project do not exist
      * @param timeout a timeout for waiting
      * @param listener the callback listener
      */
@@ -202,7 +203,8 @@ public void waitForPersistentTaskCondition(
         ClusterStateObserver.waitForState(clusterService, threadPool.getThreadContext(), new ClusterStateObserver.Listener() {
             @Override
             public void onNewClusterState(ClusterState state) {
-                listener.onResponse(PersistentTasksCustomMetadata.getTaskWithId(state.metadata().getProject(projectId), taskId));
+                final var project = state.metadata().projects().get(projectId);
+                listener.onResponse(project == null ? null : PersistentTasksCustomMetadata.getTaskWithId(project, taskId));
             }
 
             @Override
@@ -214,13 +216,15 @@ public void onClusterServiceClose() {
             public void onTimeout(TimeValue timeout) {
                 listener.onTimeout(timeout);
             }
-        },
-            clusterState -> predicate.test(
-                PersistentTasksCustomMetadata.getTaskWithId(clusterState.metadata().getProject(projectId), taskId)
-            ),
-            timeout,
-            logger
-        );
+        }, clusterState -> {
+            final var project = clusterState.metadata().projects().get(projectId);
+            if (project == null) {
+                logger.debug("project [{}] not found while waiting for persistent task [{}] to pass predicate", projectId, taskId);
+                return predicate.test(null);
+            } else {
+                return predicate.test(PersistentTasksCustomMetadata.getTaskWithId(project, taskId));
+            }
+        }, timeout, logger);
     }
 
     // visible for testing