From 99db71b486375367d796f19d21a74f373ea38a49 Mon Sep 17 00:00:00 2001
From: Aleksandr Iushmanov <aiushmanov@confluent.io>
Date: Fri, 6 Jun 2025 15:56:34 +0100
Subject: [PATCH] [FLINK-37822] Fix flaky test:
 OpenTelemetryMetricReporterITCase.testReport

---
 .../flink/metrics/otel/OpenTelemetryMetricReporter.java  | 9 ++++++++-
 .../metrics/otel/OpenTelemetryMetricReporterITCase.java  | 4 ++++
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java b/flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java
index c04be6a9bdaf6..e0e3fcea7e126 100644
--- a/flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java
+++ b/flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java
@@ -101,7 +101,7 @@ public void open(MetricConfig metricConfig) {
     @Override
     public void close() {
         exporter.flush();
-        lastResult.join(1, TimeUnit.MINUTES);
+        waitForLastReportToComplete();
         exporter.close();
     }
 
@@ -268,4 +268,11 @@ public void report() {
                     exporter.getClass().getName());
         }
     }
+
+    @VisibleForTesting
+    void waitForLastReportToComplete() {
+        if (lastResult != null) {
+            lastResult.join(1, TimeUnit.MINUTES);
+        }
+    }
 }
diff --git a/flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterITCase.java b/flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterITCase.java
index 95339c5391786..96b935fec36db 100644
--- a/flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterITCase.java
+++ b/flink-metrics/flink-metrics-otel/src/test/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporterITCase.java
@@ -80,6 +80,10 @@ public void testReport() throws Exception {
         reporter.notifyOfAddedMetric(gauge, "foo.gauge", group);
 
         reporter.report();
+        // Reporting is async, hence reports order is not deterministic.
+        // As test utils rely on verification of "last" report, we enforce report order
+        // for testing purposes.
+        reporter.waitForLastReportToComplete();
 
         MeterView meter = new MeterView(counter);
         reporter.notifyOfAddedMetric(meter, "foo.meter", group);