jmx-metrics: thoroughly wait for cassandra and kafka test containers (#219)

Author: rmfitzpatrick

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/AbstractIntegrationTest.java

@@ -112,6 +112,19 @@ void beforeEach() {
     otlpServer.reset();
   }
 
+  protected static GenericContainer<?> cassandraContainer() {
+    return new GenericContainer<>("cassandra:3.11")
+        .withNetwork(Network.SHARED)
+        .withEnv("LOCAL_JMX", "no")
+        .withCopyFileToContainer(
+            MountableFile.forClasspathResource("cassandra/jmxremote.password", 0400),
+            "/etc/cassandra/jmxremote.password")
+        .withNetworkAliases("cassandra")
+        .withExposedPorts(7199)
+        .withStartupTimeout(Duration.ofMinutes(2))
+        .waitingFor(Wait.forLogMessage(".*Startup complete.*", 1));
+  }
+
   @SafeVarargs
   protected final void waitAndAssertMetrics(Consumer<Metric>... assertions) {
     await()

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/OtlpIntegrationTest.java

@@ -9,31 +9,16 @@
 
 import io.opentelemetry.proto.common.v1.AnyValue;
 import io.opentelemetry.proto.common.v1.KeyValue;
-import java.time.Duration;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
-import org.testcontainers.utility.MountableFile;
 
 abstract class OtlpIntegrationTest extends AbstractIntegrationTest {
   OtlpIntegrationTest(boolean configFromStdin) {
     super(configFromStdin, "otlp_config.properties");
   }
 
-  @Container
-  GenericContainer<?> cassandra =
-      new GenericContainer<>("cassandra:3.11")
-          .withNetwork(Network.SHARED)
-          .withEnv("LOCAL_JMX", "no")
-          .withCopyFileToContainer(
-              MountableFile.forClasspathResource("cassandra/jmxremote.password", 0400),
-              "/etc/cassandra/jmxremote.password")
-          .withNetworkAliases("cassandra")
-          .withExposedPorts(7199)
-          .withStartupTimeout(Duration.ofMinutes(2))
-          .waitingFor(Wait.forListeningPort());
+  @Container GenericContainer<?> cassandra = cassandraContainer();
 
   @Test
   void endToEnd() {

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/target_systems/CassandraIntegrationTest.java

@@ -8,36 +8,21 @@
 import static org.assertj.core.api.Assertions.entry;
 
 import io.opentelemetry.contrib.jmxmetrics.AbstractIntegrationTest;
-import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
 import org.assertj.core.api.MapAssert;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
-import org.testcontainers.utility.MountableFile;
 
 class CassandraIntegrationTest extends AbstractIntegrationTest {
 
   CassandraIntegrationTest() {
     super(/* configFromStdin= */ false, "target-systems/cassandra.properties");
   }
 
-  @Container
-  GenericContainer<?> cassandra =
-      new GenericContainer<>("cassandra:3.11")
-          .withNetwork(Network.SHARED)
-          .withEnv("LOCAL_JMX", "no")
-          .withCopyFileToContainer(
-              MountableFile.forClasspathResource("cassandra/jmxremote.password", 0400),
-              "/etc/cassandra/jmxremote.password")
-          .withNetworkAliases("cassandra")
-          .withExposedPorts(7199)
-          .withStartupTimeout(Duration.ofMinutes(2))
-          .waitingFor(Wait.forListeningPort());
+  @Container GenericContainer<?> cassandra = cassandraContainer();
 
   @Test
   void endToEnd() {

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/target_systems/JvmTargetSystemIntegrationTest.java

@@ -6,34 +6,19 @@
 package io.opentelemetry.contrib.jmxmetrics.target_systems;
 
 import io.opentelemetry.contrib.jmxmetrics.AbstractIntegrationTest;
-import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
-import org.testcontainers.utility.MountableFile;
 
 class JvmTargetSystemIntegrationTest extends AbstractIntegrationTest {
 
   JvmTargetSystemIntegrationTest() {
     super(/* configFromStdin= */ false, "target-systems/jvm.properties");
   }
 
-  @Container
-  GenericContainer<?> cassandra =
-      new GenericContainer<>("cassandra:3.11")
-          .withNetwork(Network.SHARED)
-          .withEnv("LOCAL_JMX", "no")
-          .withCopyFileToContainer(
-              MountableFile.forClasspathResource("cassandra/jmxremote.password", 0400),
-              "/etc/cassandra/jmxremote.password")
-          .withNetworkAliases("cassandra")
-          .withExposedPorts(7199)
-          .withStartupTimeout(Duration.ofMinutes(2))
-          .waitingFor(Wait.forListeningPort());
+  @Container GenericContainer<?> cassandra = cassandraContainer();
 
   @Test
   void endToEnd() {

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/target_systems/KafkaIntegrationTest.java

@@ -50,7 +50,8 @@ protected KafkaIntegrationTest(String configName) {
           .withNetworkAliases("kafka")
           .withExposedPorts(7199)
           .withStartupTimeout(Duration.ofMinutes(2))
-          .waitingFor(Wait.forListeningPort())
+          .waitingFor(
+              Wait.forLogMessage(".*KafkaServer.*started \\(kafka.server.KafkaServer\\).*", 1))
           .dependsOn(zookeeper);
 
   Startable createTopics =
@@ -200,7 +201,7 @@ static class KafkaConsumerIntegrationTest extends KafkaIntegrationTest {
                 "--max-messages",
                 "100")
             .withStartupTimeout(Duration.ofMinutes(2))
-            .waitingFor(Wait.forListeningPort())
+            .waitingFor(Wait.forLogMessage(".*Welcome to the Bitnami kafka container.*", 1))
             .dependsOn(createTopics);
 
     @Test
@@ -281,7 +282,7 @@ static class KafkaProducerIntegrationTest extends KafkaIntegrationTest {
             .withCommand("kafka-producer.sh")
             .withStartupTimeout(Duration.ofMinutes(2))
             .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("kafka-producer")))
-            .waitingFor(Wait.forListeningPort())
+            .waitingFor(Wait.forLogMessage(".*Welcome to the Bitnami kafka container.*", 1))
             .dependsOn(createTopics);
 
     @Test