Finish monitor -- use mockito spy to check that the new monitor is called (and thus that it is set up properly).

Author: DiannaHohensee

server/src/main/java/org/elasticsearch/cluster/routing/allocation/NodeUsageStatsForThreadPoolsMonitor.java

@@ -75,6 +75,8 @@ public void onNewInfo(ClusterInfo clusterInfo) {
         String explanation = "";
         final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
 
+        // TODO (ES-11992): implement
+
         if (reroute) {
             logger.debug("rerouting shards: [{}]", explanation);
             rerouteService.reroute("disk threshold monitor", Priority.NORMAL, ActionListener.wrap(ignored -> {

server/src/test/java/org/elasticsearch/cluster/InternalClusterInfoServiceSchedulingTests.java

@@ -21,11 +21,14 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.routing.RerouteService;
+import org.elasticsearch.cluster.routing.allocation.NodeUsageStatsForThreadPoolsMonitor;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintSettings;
 import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.FakeThreadPoolMasterService;
 import org.elasticsearch.cluster.service.MasterService;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
@@ -95,6 +98,18 @@ protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
             mockEstimatedHeapUsageCollector,
             nodeUsageStatsForThreadPoolsCollector
         );
+        final NodeUsageStatsForThreadPoolsMonitor usageMonitor = spy(
+            new NodeUsageStatsForThreadPoolsMonitor(
+                clusterService.getClusterSettings(),
+                threadPool.relativeTimeInMillisSupplier(),
+                clusterService::state,
+                new RerouteService() {
+                    @Override
+                    public void reroute(String reason, Priority priority, ActionListener<Void> listener) {}
+                }
+            )
+        );
+        clusterInfoService.addListener(usageMonitor::onNewInfo);
         clusterService.addListener(clusterInfoService);
         clusterInfoService.addListener(ignored -> {});
 
@@ -132,13 +147,15 @@ protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
         for (int i = 0; i < 3; i++) {
             Mockito.clearInvocations(mockEstimatedHeapUsageCollector);
             Mockito.clearInvocations(nodeUsageStatsForThreadPoolsCollector);
+            Mockito.clearInvocations(usageMonitor);
             final int initialRequestCount = client.requestCount;
             final long duration = INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING.get(settings).millis();
             runFor(deterministicTaskQueue, duration);
             deterministicTaskQueue.runAllRunnableTasks();
             assertThat(client.requestCount, equalTo(initialRequestCount + 2)); // should have run two client requests per interval
             verify(mockEstimatedHeapUsageCollector).collectClusterHeapUsage(any()); // Should poll for heap usage once per interval
             verify(nodeUsageStatsForThreadPoolsCollector).collectUsageStats(any(), any(), any());
+            verify(usageMonitor).onNewInfo(any());
         }
 
         final AtomicBoolean failMaster2 = new AtomicBoolean();