Propagate flush to PeriodicMetricReader's metricExporter. (#7410)

Author: tylerbenson

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

@@ -81,7 +81,21 @@ public MemoryMode getMemoryMode() {
 
   @Override
   public CompletableResultCode forceFlush() {
-    return scheduled.doRun();
+    CompletableResultCode result = new CompletableResultCode();
+    CompletableResultCode doRunResult = scheduled.doRun();
+    doRunResult.whenComplete(
+        () -> {
+          CompletableResultCode flushResult = exporter.flush();
+          flushResult.whenComplete(
+              () -> {
+                if (doRunResult.isSuccess() && flushResult.isSuccess()) {
+                  result.succeed();
+                } else {
+                  result.fail();
+                }
+              });
+        });
+    return result;
   }
 
   @Override

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

@@ -154,6 +154,35 @@ void flush() throws Exception {
     }
   }
 
+  @Test
+  void forceflush_callsFlush() {
+    MetricExporter metricExporter = mock(MetricExporter.class);
+    when(metricExporter.export(any()))
+        .thenReturn(CompletableResultCode.ofSuccess())
+        .thenReturn(CompletableResultCode.ofSuccess())
+        .thenThrow(new RuntimeException("Export Failed!"));
+    when(metricExporter.flush())
+        .thenReturn(CompletableResultCode.ofSuccess())
+        .thenReturn(CompletableResultCode.ofFailure())
+        .thenReturn(CompletableResultCode.ofSuccess());
+    when(metricExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
+
+    PeriodicMetricReader reader =
+        PeriodicMetricReader.builder(metricExporter)
+            .setInterval(Duration.ofNanos(Long.MAX_VALUE))
+            .build();
+
+    try {
+      reader.register(collectionRegistration);
+      assertThat(reader.forceFlush().join(10, TimeUnit.SECONDS).isSuccess()).isTrue();
+      assertThat(reader.forceFlush().join(10, TimeUnit.SECONDS).isSuccess()).isFalse();
+      assertThat(reader.forceFlush().join(10, TimeUnit.SECONDS).isSuccess()).isFalse();
+    } finally {
+      reader.shutdown();
+    }
+    verify(metricExporter, times(3)).flush();
+  }
+
   @Test
   @Timeout(2)
   @SuppressLogger(PeriodicMetricReader.class)