Assert latch.await returns true and always return from listener

Author: pickypg

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporter.java

@@ -677,6 +677,9 @@ public void openBulk(final ActionListener<ExportBulk> listener) {
                 final String name = "xpack.monitoring.exporters." + config.name();
 
                 listener.onResponse(new HttpExportBulk(name, client, defaultParams, dateTimeFormatter, threadContext));
+            } else {
+                // we're not ready yet, so keep waiting
+                listener.onResponse(null);
             }
         }, listener::onFailure));
     }

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporterIT.java

@@ -295,7 +295,7 @@ public void testHostChangeReChecksTemplate() throws Exception {
     }
 
     public void testUnsupportedClusterVersion() throws Exception {
-        Settings settings = Settings.builder()
+        final Settings settings = Settings.builder()
                 .put("xpack.monitoring.exporters._http.type", "http")
                 .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                 .build();
@@ -317,15 +317,25 @@ public void testUnsupportedClusterVersion() throws Exception {
                 bulk -> {
                     assertNull(bulk);
 
+                    logger.error("[testUnsupportedClusterVersion] responded to listener");
+
                     awaitResponseAndClose.countDown();
                 },
-                e -> fail(e.getMessage())
+                e -> {
+                    logger.error("[testUnsupportedClusterVersion] failed to respond to listener");
+
+                    fail(e.getMessage());
+                }
             );
 
             exporter.openBulk(listener);
 
+            logger.error("[testUnsupportedClusterVersion] waiting for latch");
+
             // wait for it to actually respond
-            awaitResponseAndClose.await(15, TimeUnit.SECONDS);
+            assertTrue(awaitResponseAndClose.await(15, TimeUnit.SECONDS));
+
+            logger.error("[testUnsupportedClusterVersion] waited for latch");
         }
 
         assertThat(webServer.requests(), hasSize(1));
@@ -593,7 +603,7 @@ private void export(final Settings settings, final Collection<MonitoringDoc> doc
             }, e -> fail("Failed to create HttpExportBulk")));
 
             // block until the bulk responds
-            awaitResponseAndClose.await(15, TimeUnit.SECONDS);
+            assertTrue(awaitResponseAndClose.await(15, TimeUnit.SECONDS));
         }
     }
 

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporterTests.java

@@ -453,7 +453,35 @@ public void testHttpExporterDirtyResourcesBlock() throws Exception {
             exporter.openBulk(bulkListener);
 
             // wait for it to actually respond
-            awaitResponseAndClose.await(15, TimeUnit.SECONDS);
+            assertTrue(awaitResponseAndClose.await(15, TimeUnit.SECONDS));
+        }
+    }
+
+    public void testHttpExporterReturnsNullForOpenBulkIfNotReady() throws Exception {
+        final Config config = createConfig(Settings.EMPTY);
+        final RestClient client = mock(RestClient.class);
+        final Sniffer sniffer = randomFrom(mock(Sniffer.class), null);
+        final NodeFailureListener listener = mock(NodeFailureListener.class);
+        // always has to check, and never succeeds checks but it does not throw an exception (e.g., version check fails)
+        final HttpResource resource = new MockHttpResource(exporterName(), true, false, false);
+
+        try (HttpExporter exporter = new HttpExporter(config, client, sniffer, threadContext, listener, resource)) {
+            verify(listener).setResource(resource);
+
+            final CountDownLatch awaitResponseAndClose = new CountDownLatch(1);
+            final ActionListener<ExportBulk> bulkListener = ActionListener.wrap(
+                bulk -> {
+                    assertThat(bulk, nullValue());
+
+                    awaitResponseAndClose.countDown();
+                },
+                e -> fail(e.getMessage())
+            );
+
+            exporter.openBulk(bulkListener);
+
+            // wait for it to actually respond
+            assertTrue(awaitResponseAndClose.await(15, TimeUnit.SECONDS));
         }
     }
 
@@ -481,7 +509,7 @@ public void testHttpExporter() throws Exception {
             exporter.openBulk(bulkListener);
 
             // wait for it to actually respond
-            awaitResponseAndClose.await(15, TimeUnit.SECONDS);
+            assertTrue(awaitResponseAndClose.await(15, TimeUnit.SECONDS));
         }
     }
 

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpResourceTests.java

@@ -146,7 +146,7 @@ protected void doCheckAndPublish(RestClient client, ActionListener<Boolean> list
                 // then respond here
                 final Thread thread = new Thread(() -> {
                     try {
-                        secondCheck.await(15, TimeUnit.SECONDS);
+                        assertTrue(secondCheck.await(15, TimeUnit.SECONDS));
                         listener.onResponse(response);
                     } catch (InterruptedException e) {
                         listener.onFailure(e);
@@ -161,7 +161,7 @@ protected void doCheckAndPublish(RestClient client, ActionListener<Boolean> list
         resource.checkAndPublishIfDirty(client, listener);
         resource.checkAndPublishIfDirty(client, checkingListener);
 
-        firstCheck.await(15, TimeUnit.SECONDS);
+        assertTrue(firstCheck.await(15, TimeUnit.SECONDS));
 
         verify(listener).onResponse(response);