Fix jmx-metrics kafka test (#1822)

Author: laurit

jmx-metrics/build.gradle.kts

@@ -50,6 +50,7 @@ testing {
         implementation("com.linecorp.armeria:armeria-junit5")
         implementation("io.opentelemetry.proto:opentelemetry-proto:1.5.0-alpha")
         implementation("org.testcontainers:junit-jupiter")
+        implementation("org.slf4j:slf4j-simple")
       }
     }
   }

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

@@ -32,6 +32,7 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import org.assertj.core.api.MapAssert;
+import org.awaitility.core.ConditionTimeoutException;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -134,28 +135,41 @@ protected static GenericContainer<?> cassandraContainer() {
   }
 
   protected final void waitAndAssertMetrics(Iterable<Consumer<Metric>> assertions) {
-    await()
-        .atMost(Duration.ofSeconds(30))
-        .untilAsserted(
-            () -> {
-              List<Metric> metrics =
-                  otlpServer.getMetrics().stream()
-                      .map(ExportMetricsServiceRequest::getResourceMetricsList)
-                      .flatMap(rm -> rm.stream().map(ResourceMetrics::getScopeMetricsList))
-                      .flatMap(Collection::stream)
-                      .filter(
-                          sm ->
-                              sm.getScope().getName().equals("io.opentelemetry.contrib.jmxmetrics")
-                                  && sm.getScope().getVersion().equals(expectedMeterVersion()))
-                      .flatMap(sm -> sm.getMetricsList().stream())
-                      .collect(Collectors.toList());
-
-              assertThat(metrics).isNotEmpty();
-
-              for (Consumer<Metric> assertion : assertions) {
-                assertThat(metrics).anySatisfy(assertion);
-              }
-            });
+    waitAndAssertMetrics(
+        () -> {
+          List<Metric> metrics =
+              otlpServer.getMetrics().stream()
+                  .map(ExportMetricsServiceRequest::getResourceMetricsList)
+                  .flatMap(rm -> rm.stream().map(ResourceMetrics::getScopeMetricsList))
+                  .flatMap(Collection::stream)
+                  .filter(
+                      sm ->
+                          sm.getScope().getName().equals("io.opentelemetry.contrib.jmxmetrics")
+                              && sm.getScope().getVersion().equals(expectedMeterVersion()))
+                  .flatMap(sm -> sm.getMetricsList().stream())
+                  .collect(Collectors.toList());
+
+          assertThat(metrics).isNotEmpty();
+
+          for (Consumer<Metric> assertion : assertions) {
+            assertThat(metrics).anySatisfy(assertion);
+          }
+        });
+  }
+
+  private static void waitAndAssertMetrics(Runnable assertion) {
+    try {
+      await().atMost(Duration.ofSeconds(30)).untilAsserted(assertion::run);
+    } catch (Throwable t) {
+      if (t instanceof ConditionTimeoutException) {
+        // Don't throw this failure since the stack is the awaitility thread, causing confusion.
+        // Instead, just assert one more time on the test thread, which will fail with a better
+        // stack trace - that is on the same thread as the test.
+        assertion.run();
+      } else {
+        throw t;
+      }
+    }
   }
 
   @SafeVarargs

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

@@ -80,7 +80,7 @@ public Set<Startable> getDependencies() {
       };
 
   protected GenericContainer<?> kafkaProducerContainer() {
-    return new GenericContainer<>("bitnami/kafka:latest")
+    return new GenericContainer<>("bitnami/kafka:2.8.1")
         .withNetwork(Network.SHARED)
         .withEnv("KAFKA_CFG_ZOOKEEPER_CONNECT", "zookeeper:2181")
         .withEnv("ALLOW_PLAINTEXT_LISTENER", "yes")
@@ -207,7 +207,7 @@ static class KafkaConsumerIntegrationTest extends KafkaIntegrationTest {
 
     @Container
     GenericContainer<?> consumer =
-        new GenericContainer<>("bitnami/kafka:latest")
+        new GenericContainer<>("bitnami/kafka:2.8.1")
             .withNetwork(Network.SHARED)
             .withEnv("KAFKA_CFG_ZOOKEEPER_CONNECT", "zookeeper:2181")
             .withEnv("ALLOW_PLAINTEXT_LISTENER", "yes")