Remove LegacyCTRAListener usage in MasterService (#83796)

Relates #83784

server/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -112,7 +112,7 @@ public class TransportClusterHealthAction extends TransportMasterNodeReadAction<
         if (request.local()) {
             new LocalMasterServiceTask(request.waitForEvents()) {
                 @Override
-                public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
+                protected void onPublicationComplete() {
                     final long timeoutInMillis = Math.max(0, endTimeRelativeMillis - threadPool.relativeTimeInMillis());
                     final TimeValue newTimeout = TimeValue.timeValueMillis(timeoutInMillis);
                     request.timeout(newTimeout);

server/src/main/java/org/elasticsearch/cluster/ClusterStateTaskExecutor.java

@@ -157,9 +157,17 @@ public interface ClusterStateTaskExecutor<T extends ClusterStateTaskListener> {
                 assert tasks.size() == 1 : "this only supports a single task but received " + tasks;
                 final T task = tasks.get(0);
                 final ClusterState newState = task.execute(currentState);
-                return ClusterTasksResult.<T>builder()
-                    .success(task, new LegacyClusterTaskResultActionListener(task, currentState))
-                    .build(newState);
+                return ClusterTasksResult.<T>builder().success(task, new ActionListener<>() {
+                    @Override
+                    public void onResponse(ClusterState publishedState) {
+                        task.clusterStateProcessed(currentState, publishedState);
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        task.onFailure(e);
+                    }
+                }).build(newState);
             }
 
             @Override

server/src/main/java/org/elasticsearch/cluster/LocalMasterServiceTask.java

@@ -7,6 +7,7 @@
  */
 package org.elasticsearch.cluster;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Priority;
 
@@ -23,7 +24,14 @@ public abstract class LocalMasterServiceTask implements ClusterStateTaskListener
         this.priority = priority;
     }
 
-    public void execute(ClusterState currentState) throws Exception {}
+    protected void execute(ClusterState currentState) throws Exception {}
+
+    @Override
+    public final void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
+        assert false : "not called";
+    }
+
+    protected void onPublicationComplete() {}
 
     public void submit(MasterService masterService, String source) {
         masterService.submitStateUpdateTask(
@@ -51,9 +59,17 @@ public abstract class LocalMasterServiceTask implements ClusterStateTaskListener
                     assert tasks.size() == 1 && tasks.get(0) == thisTask
                         : "expected one-element task list containing current object but was " + tasks;
                     thisTask.execute(currentState);
-                    return ClusterTasksResult.<LocalMasterServiceTask>builder()
-                        .success(thisTask, new LegacyClusterTaskResultActionListener(thisTask, currentState))
-                        .build(currentState);
+                    return ClusterTasksResult.<LocalMasterServiceTask>builder().success(thisTask, new ActionListener<>() {
+                        @Override
+                        public void onResponse(ClusterState clusterState) {
+                            onPublicationComplete();
+                        }
+
+                        @Override
+                        public void onFailure(Exception e) {
+                            LocalMasterServiceTask.this.onFailure(e);
+                        }
+                    }).build(currentState);
                 }
             }
         );