Internal: Readd accidental removed functionality in InternalClusterSe…

…rvice

The commit about adding cluster health response features also removed
accidentally some functionality, that resulted in wrong instanceof checks
in InternalClusterService and thus in test failures because the cluster
state task that was added via an anonymous was missing the cast.

This commit readds the abstract class with slight renaming.

Commit id was: 88f8d58

core/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java

@@ -243,7 +243,7 @@ public void add(@Nullable final TimeValue timeout, final TimeoutClusterStateList
         }
         // call the post added notification on the same event thread
         try {
-            updateTasksExecutor.execute(new PrioritizedRunnable(Priority.HIGH) {
+            updateTasksExecutor.execute(new SourcePrioritizedRunnable(Priority.HIGH, "_add_listener_") {
                 @Override
                 public void run() {
                     if (timeout != null) {
@@ -312,12 +312,12 @@ public List<PendingClusterTask> pendingTasks() {
             final Object task = pending.task;
             if (task == null) {
                 continue;
-            } else if (task instanceof UpdateTask) {
-                UpdateTask runnable = (UpdateTask) task;
+            } else if (task instanceof SourcePrioritizedRunnable) {
+                SourcePrioritizedRunnable runnable = (SourcePrioritizedRunnable) task;
                 source = runnable.source();
                 timeInQueue = runnable.getAgeInMillis();
             } else {
-                assert false : "expected TimedPrioritizedRunnable got " + task.getClass();
+                assert false : "expected SourcePrioritizedRunnable got " + task.getClass();
                 source = "unknown [" + task.getClass() + "]";
                 timeInQueue = 0;
             }
@@ -337,21 +337,27 @@ public TimeValue getMaxTaskWaitTime() {
         return updateTasksExecutor.getMaxTaskWaitTime();
     }
 
-    class UpdateTask extends PrioritizedRunnable {
-
-        public final ClusterStateUpdateTask updateTask;
+    static abstract class SourcePrioritizedRunnable extends PrioritizedRunnable {
         protected final String source;
 
-
-        UpdateTask(String source, Priority priority, ClusterStateUpdateTask updateTask) {
+        public SourcePrioritizedRunnable(Priority priority, String source) {
             super(priority);
-            this.updateTask = updateTask;
             this.source = source;
         }
 
         public String source() {
             return source;
         }
+    }
+
+    class UpdateTask extends SourcePrioritizedRunnable {
+
+        public final ClusterStateUpdateTask updateTask;
+
+        UpdateTask(String source, Priority priority, ClusterStateUpdateTask updateTask) {
+            super(priority, source);
+            this.updateTask = updateTask;
+        }
 
         @Override
         public void run() {