Clarify closing of timeout listeners

Author: jasontedor

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

@@ -219,8 +219,12 @@ synchronized protected void doStop() {
             }
         }
         ThreadPool.terminate(updateTasksExecutor, 10, TimeUnit.SECONDS);
-        postAppliedListeners.stream().filter(listener -> listener instanceof TimeoutClusterStateListener)
-                .forEach(listener -> ((TimeoutClusterStateListener) listener).onClose());
+        // close timeout listeners that did not have an ongoing timeout
+        postAppliedListeners
+                .stream()
+                .filter(listener -> listener instanceof TimeoutClusterStateListener)
+                .map(listener -> (TimeoutClusterStateListener)listener)
+                .forEach(TimeoutClusterStateListener::onClose);
         remove(localNodeMasterListeners);
     }