ForceFlush Implementation

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsCustomizerProvider.java

@@ -17,7 +17,6 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
-import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.contrib.awsxray.AlwaysRecordSampler;
 import io.opentelemetry.contrib.awsxray.ResourceHolder;
 import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
@@ -164,14 +163,16 @@ private SdkTracerProviderBuilder customizeTracerProviderBuilder(
       MetricReader metricReader =
           PeriodicMetricReader.builder(metricsExporter).setInterval(exportInterval).build();
 
-      MeterProvider meterProvider =
+      SdkMeterProvider meterProvider =
           SdkMeterProvider.builder()
               .setResource(ResourceHolder.getResource())
               .registerMetricReader(metricReader)
               .build();
+
       // Construct and set application signals metrics processor
       SpanProcessor spanMetricsProcessor =
-          AwsSpanMetricsProcessorBuilder.create(meterProvider, ResourceHolder.getResource())
+          AwsSpanMetricsProcessorBuilder.create(
+                  meterProvider, ResourceHolder.getResource(), meterProvider::forceFlush)
               .build();
       tracerProviderBuilder.addSpanProcessor(spanMetricsProcessor);
     }

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessor.java

@@ -22,12 +22,14 @@
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.ReadWriteSpan;
 import io.opentelemetry.sdk.trace.ReadableSpan;
 import io.opentelemetry.sdk.trace.SpanProcessor;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.util.Map;
+import java.util.function.Supplier;
 import javax.annotation.concurrent.Immutable;
 
 /**
@@ -64,29 +66,38 @@ public final class AwsSpanMetricsProcessor implements SpanProcessor {
 
   private final MetricAttributeGenerator generator;
   private final Resource resource;
+  private final Supplier<CompletableResultCode> forceFlushAction;
 
   /** Use {@link AwsSpanMetricsProcessorBuilder} to construct this processor. */
   static AwsSpanMetricsProcessor create(
       LongHistogram errorHistogram,
       LongHistogram faultHistogram,
       DoubleHistogram latencyHistogram,
       MetricAttributeGenerator generator,
-      Resource resource) {
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     return new AwsSpanMetricsProcessor(
-        errorHistogram, faultHistogram, latencyHistogram, generator, resource);
+        errorHistogram, faultHistogram, latencyHistogram, generator, resource, forceFlushAction);
   }
 
   private AwsSpanMetricsProcessor(
       LongHistogram errorHistogram,
       LongHistogram faultHistogram,
       DoubleHistogram latencyHistogram,
       MetricAttributeGenerator generator,
-      Resource resource) {
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     this.errorHistogram = errorHistogram;
     this.faultHistogram = faultHistogram;
     this.latencyHistogram = latencyHistogram;
     this.generator = generator;
     this.resource = resource;
+    this.forceFlushAction = forceFlushAction;
+  }
+
+  @Override
+  public CompletableResultCode forceFlush() {
+    return forceFlushAction.get();
   }
 
   @Override

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorBuilder.java

@@ -22,7 +22,9 @@
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.resources.Resource;
+import java.util.function.Supplier;
 
 /** A builder for {@link AwsSpanMetricsProcessor} */
 public final class AwsSpanMetricsProcessorBuilder {
@@ -42,18 +44,29 @@ public final class AwsSpanMetricsProcessorBuilder {
   private final MeterProvider meterProvider;
   private final Resource resource;
 
+  // ForceFlush action provided from {@link SdkMeterProvider#forceFlush()} so that when the
+  // application exits The spanMetricProcessor calls the meterProvder.forceFlush to flush
+  // any remaining metrics before shutdown
+  private final Supplier<CompletableResultCode> forceFlushAction;
+
   // Optional builder elements
   private MetricAttributeGenerator generator = DEFAULT_GENERATOR;
   private String scopeName = DEFAULT_SCOPE_NAME;
 
   public static AwsSpanMetricsProcessorBuilder create(
-      MeterProvider meterProvider, Resource resource) {
-    return new AwsSpanMetricsProcessorBuilder(meterProvider, resource);
+      MeterProvider meterProvider,
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
+    return new AwsSpanMetricsProcessorBuilder(meterProvider, resource, forceFlushAction);
   }
 
-  private AwsSpanMetricsProcessorBuilder(MeterProvider meterProvider, Resource resource) {
+  private AwsSpanMetricsProcessorBuilder(
+      MeterProvider meterProvider,
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     this.meterProvider = meterProvider;
     this.resource = resource;
+    this.forceFlushAction = forceFlushAction;
   }
 
   /**
@@ -86,6 +99,6 @@ public AwsSpanMetricsProcessor build() {
         meter.histogramBuilder(LATENCY).setUnit(LATENCY_UNITS).build();
 
     return AwsSpanMetricsProcessor.create(
-        errorHistogram, faultHistogram, latencyHistogram, generator, resource);
+        errorHistogram, faultHistogram, latencyHistogram, generator, resource, forceFlushAction);
   }
 }

awsagentprovider/src/test/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorTest.java

@@ -77,6 +77,12 @@ private enum ExpectedStatusMetric {
   private MetricAttributeGenerator generatorMock;
   private AwsSpanMetricsProcessor awsSpanMetricsProcessor;
 
+  // Mock forceFlush function that returns success when invoked similar
+  // to the default implementation of forceFlush.
+  private CompletableResultCode forceFlushAction() {
+    return CompletableResultCode.ofSuccess();
+  }
+
   @BeforeEach
   public void setUpMocks() {
     errorHistogramMock = mock(LongHistogram.class);
@@ -90,7 +96,8 @@ public void setUpMocks() {
             faultHistogramMock,
             latencyHistogramMock,
             generatorMock,
-            testResource);
+            testResource,
+            this::forceFlushAction);
   }
 
   @Test