Make persistent task name and ID available in update source (#122334)

Add the task name (when possible) and ID as part of the source string
for updating cluster state. This helps better identifying the source of
a task. The updatePersistentTaskState method already does it. This PR
ensures it is the case in other places.

Author: ywangd

server/src/internalClusterTest/java/org/elasticsearch/persistent/PersistentTaskCreationFailureIT.java

@@ -77,7 +77,9 @@ public ClusterState execute(ClusterState currentState) {
                                 .pendingTasks()
                                 .stream()
                                 .filter(
-                                    pendingClusterTask -> pendingClusterTask.getSource().string().equals("finish persistent task (failed)")
+                                    pendingClusterTask -> pendingClusterTask.getSource()
+                                        .string()
+                                        .matches("finish persistent task \\[.*] \\(failed\\)")
                                 )
                                 .count();
                             assertThat(completePersistentTaskPendingTasksCount, lessThanOrEqualTo(1L));

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

@@ -113,7 +113,7 @@ public <Params extends PersistentTaskParams> void createPersistentTask(
         Params taskParams,
         ActionListener<PersistentTask<?>> listener
     ) {
-        submitUnbatchedTask("create persistent task", new ClusterStateUpdateTask() {
+        submitUnbatchedTask("create persistent task " + taskName + " [" + taskId + "]", new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 PersistentTasksCustomMetadata.Builder builder = builder(currentState);
@@ -166,9 +166,9 @@ public void completePersistentTask(String id, long allocationId, Exception failu
         final String source;
         if (failure != null) {
             logger.warn("persistent task " + id + " failed", failure);
-            source = "finish persistent task (failed)";
+            source = "finish persistent task [" + id + "] (failed)";
         } else {
-            source = "finish persistent task (success)";
+            source = "finish persistent task [" + id + "] (success)";
         }
         submitUnbatchedTask(source, new ClusterStateUpdateTask() {
             @Override
@@ -212,7 +212,7 @@ public void clusterStateProcessed(ClusterState oldState, ClusterState newState)
      * @param listener the listener that will be called when task is removed
      */
     public void removePersistentTask(String id, ActionListener<PersistentTask<?>> listener) {
-        submitUnbatchedTask("remove persistent task", new ClusterStateUpdateTask() {
+        submitUnbatchedTask("remove persistent task [" + id + "]", new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 PersistentTasksCustomMetadata.Builder tasksInProgress = builder(currentState);
@@ -295,7 +295,7 @@ public void unassignPersistentTask(
         final String reason,
         final ActionListener<PersistentTask<?>> listener
     ) {
-        submitUnbatchedTask("unassign persistent task from any node", new ClusterStateUpdateTask() {
+        submitUnbatchedTask("unassign persistent task [" + taskId + "] from any node", new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) throws Exception {
                 PersistentTasksCustomMetadata.Builder tasksInProgress = builder(currentState);

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/action/TransportSetTransformUpgradeModeActionTests.java

@@ -38,6 +38,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.matches;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -176,7 +177,7 @@ public void testEnableUpgradeMode() throws InterruptedException {
 
         upgradeModeSuccessfullyChanged(stateWithTransformTask(), assertNoFailureListener(r -> {
             assertThat(r, is(AcknowledgedResponse.TRUE));
-            verify(clusterService).submitUnbatchedStateUpdateTask(eq("unassign persistent task from any node"), any());
+            verify(clusterService).submitUnbatchedStateUpdateTask(matches("unassign persistent task \\[.*\\] from any node"), any());
         }));
     }