[FLINK-40069][tests] Stabilize OpenTelemetryMetricReporterProtocolTest against collector startup race

flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterProtocolTest.java

@@ -57,13 +57,15 @@ protected void setupAndReport(MetricConfig config) {
         reporter.open(config);
         SimpleCounter counter = new SimpleCounter();
         reporter.notifyOfAddedMetric(counter, TEST_METRIC_NAME, metricGroup);
-        reporter.report();
-        reporter.waitForLastReportToComplete();
     }
 
     @Override
     protected void assertReported() throws Exception {
         eventuallyConsumeJson(
+                () -> {
+                    reporter.report();
+                    reporter.waitForLastReportToComplete();
+                },
                 json -> assertThat(extractMetricNames(json)).contains(EXPECTED_METRIC_NAME));
     }
 

flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryTestBase.java

@@ -90,8 +90,20 @@ public static MetricConfig createMetricConfig() {
 
     public static void eventuallyConsumeJson(ThrowingConsumer<JsonNode, Exception> jsonConsumer)
             throws Exception {
+        eventuallyConsumeJson(() -> {}, jsonConsumer);
+    }
+
+    /**
+     * Runs {@code beforeAttempt} at the start of every retry iteration, so on-demand reporters can
+     * re-export instead of letting a single failed export doom the whole retry budget.
+     */
+    public static void eventuallyConsumeJson(
+            ThrowingRunnable<Exception> beforeAttempt,
+            ThrowingConsumer<JsonNode, Exception> jsonConsumer)
+            throws Exception {
         eventually(
                 () -> {
+                    beforeAttempt.run();
                     // otel-collector dumps every report in a new line, so in order to re-use the
                     // same collector across multiple tests, let's read only the last line
                     getOtelContainer()

flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OtelTestContainer.java

@@ -20,13 +20,15 @@
 
 import com.github.dockerjava.api.command.InspectContainerResponse;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.utility.Base58;
 import org.testcontainers.utility.DockerImageName;
 import org.testcontainers.utility.MountableFile;
 
 import java.io.File;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Duration;
 import java.util.Locale;
 
 /** {@link OtelTestContainer} provides an {@code Otel} test instance. */
@@ -53,6 +55,14 @@ public OtelTestContainer(Path outputDir) {
                         new File(outputDir.toFile(), LOG_FILE).getAbsolutePath(), 755),
                 getOutputLogPath().toString());
         withCommand("--config", "otel-config.yaml");
+        // The default HostPortWaitStrategy reports this shell-less image as ready before the OTLP
+        // receiver accepts connections; the readiness log line is only emitted after all
+        // components, including the receivers, have started.
+        waitingFor(
+                Wait.forLogMessage(
+                                ".*Everything is ready\\. Begin running and processing data\\..*",
+                                1)
+                        .withStartupTimeout(Duration.ofMinutes(1)));
     }
 
     public Path getOutputLogPath() {