Switch to sequential export

Author: psx95

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java

@@ -16,9 +16,8 @@
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
+import java.util.Iterator;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -226,15 +225,44 @@ CompletableResultCode doRun() {
             flushResult.succeed();
             exportAvailable.set(true);
           } else {
-            Collection<Collection<MetricData>> batches = null;
             CompletableResultCode result;
             if (metricsBatcher != null) {
-              batches = metricsBatcher.batchMetrics(metricData);
-              List<CompletableResultCode> results = new ArrayList<>(batches.size());
-              for (Collection<MetricData> batch : batches) {
-                results.add(exporter.export(batch));
-              }
-              result = CompletableResultCode.ofAll(results);
+              Collection<Collection<MetricData>> batches = metricsBatcher.batchMetrics(metricData);
+              CompletableResultCode sequentialResult = new CompletableResultCode();
+              AtomicBoolean anyFailed = new AtomicBoolean(false);
+              Iterator<Collection<MetricData>> batchIterator = batches.iterator();
+
+              Runnable exportNext =
+                  new Runnable() {
+                    @Override
+                    public void run() {
+                      while (batchIterator.hasNext()) {
+                        Collection<MetricData> currentBatch = batchIterator.next();
+                        CompletableResultCode currentResult = exporter.export(currentBatch);
+                        if (currentResult.isDone()) {
+                          if (!currentResult.isSuccess()) {
+                            anyFailed.set(true);
+                          }
+                        } else {
+                          currentResult.whenComplete(
+                              () -> {
+                                if (!currentResult.isSuccess()) {
+                                  anyFailed.set(true);
+                                }
+                                this.run();
+                              });
+                          return;
+                        }
+                      }
+                      if (anyFailed.get()) {
+                        sequentialResult.fail();
+                      } else {
+                        sequentialResult.succeed();
+                      }
+                    }
+                  };
+              exportNext.run();
+              result = sequentialResult;
             } else {
               result = exporter.export(metricData);
             }

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderTest.java

@@ -451,6 +451,48 @@ void invalidConfig() {
         .hasMessage("executor");
   }
 
+  @Test
+  void periodicExport_SequentialBatches() throws Exception {
+    MetricExporter mockExporter = mock(MetricExporter.class);
+    when(mockExporter.getAggregationTemporality(any()))
+        .thenReturn(AggregationTemporality.CUMULATIVE);
+    when(mockExporter.flush()).thenReturn(CompletableResultCode.ofSuccess());
+    when(mockExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
+
+    CompletableResultCode batch1Result = new CompletableResultCode();
+    CompletableResultCode batch2Result = CompletableResultCode.ofSuccess();
+
+    // Configure mock to return pending for 1st call, success for 2nd
+    when(mockExporter.export(any())).thenReturn(batch1Result).thenReturn(batch2Result);
+
+    PeriodicMetricReader reader =
+        PeriodicMetricReader.builder(mockExporter)
+            .setInterval(
+                Duration.ofSeconds(Integer.MAX_VALUE)) // Long interval to prevent auto-trigger
+            .setMaxExportBatchSize(3)
+            .build();
+    // Setup metrics that will result in 2 batches (we have 6 points in
+    // LONG_POINT_LIST)
+    when(collectionRegistration.collectAllMetrics())
+        .thenReturn(Collections.singletonList(METRIC_DATA));
+    reader.register(collectionRegistration);
+
+    // Trigger manual flush
+    CompletableResultCode flushResult = reader.forceFlush();
+    // Verify that the first batch WAS exported
+    verify(mockExporter, times(1)).export(any());
+    // At this point, batch 1 is stuck waiting. Batch 2 should NOT be exported yet.
+    // We verify that export was only called once in total so far.
+    verify(mockExporter, times(1)).export(any());
+    // Now we complete the first batch
+    batch1Result.succeed();
+    // Verify that the second batch IS NOW exported
+    verify(mockExporter, times(2)).export(any());
+    // Ensure the flush operation completes successfully
+    assertThat(flushResult.join(5, TimeUnit.SECONDS).isSuccess()).isTrue();
+    reader.shutdown();
+  }
+
   @Test
   void stringRepresentation() {
     when(metricExporter.toString()).thenReturn("MockMetricExporter{}");