Issue 8831 exporter backpressure

docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt

@@ -1,2 +1,4 @@
 Comparing source compatibility of opentelemetry-sdk-logs-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-logs-1.67.0.jar
-No changes.
+***  MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.logs.export.LogRecordExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) int getConcurrencyLimit()

docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt

@@ -1,2 +1,4 @@
 Comparing source compatibility of opentelemetry-sdk-metrics-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.67.0.jar
-No changes.
+***  MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.metrics.export.MetricExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) int getConcurrencyLimit()

docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt

@@ -1,2 +1,4 @@
 Comparing source compatibility of opentelemetry-sdk-trace-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-trace-1.67.0.jar
-No changes.
+***  MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.trace.export.SpanExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) int getConcurrencyLimit()

sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java

@@ -60,6 +60,7 @@ class TracerProviderConfigurationTest {
   @BeforeEach
   void setUp() {
     when(mockSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
+    when(mockSpanExporter.getConcurrencyLimit()).thenReturn(1);
   }
 
   @Test

sdk/all/src/test/java/io/opentelemetry/sdk/OpenTelemetrySdkTest.java

@@ -215,6 +215,7 @@ void getMeter() {
 
   @Test
   void meterBuilder() {
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     OpenTelemetrySdk openTelemetry =
         OpenTelemetrySdk.builder()
@@ -229,6 +230,7 @@ void meterBuilder() {
 
   @Test
   void meterBuilder_ViaProvider() {
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     OpenTelemetrySdk openTelemetry =
         OpenTelemetrySdk.builder()
@@ -257,6 +259,7 @@ void getMeterProvider() {
   // Demonstrates how clear or confusing is SDK configuration
   @Test
   void fullOpenTelemetrySdkConfigurationDemo() {
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     OpenTelemetrySdk.builder()
         .setMeterProvider(
@@ -294,6 +297,7 @@ void fullOpenTelemetrySdkConfigurationDemo() {
   // Demonstrates how clear or confusing is SDK configuration
   @Test
   void trivialOpenTelemetrySdkConfigurationDemo() {
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     OpenTelemetrySdk.builder()
         .setMeterProvider(
@@ -312,6 +316,7 @@ void trivialOpenTelemetrySdkConfigurationDemo() {
   // Demonstrates how clear or confusing is SDK configuration
   @Test
   void minimalOpenTelemetrySdkConfigurationDemo() {
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     OpenTelemetrySdk.builder()
         .setMeterProvider(
@@ -384,6 +389,7 @@ void shutdown() {
   void stringRepresentation() {
     SpanExporter spanExporter = mock(SpanExporter.class);
     when(spanExporter.toString()).thenReturn("MockSpanExporter{}");
+    when(metricExporter.getConcurrencyLimit()).thenReturn(1);
     when(metricExporter.getDefaultAggregation(any())).thenCallRealMethod();
     when(metricExporter.toString()).thenReturn("MockMetricExporter{}");
     LogRecordExporter logRecordExporter = mock(LogRecordExporter.class);

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java

@@ -19,15 +19,21 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.annotation.Nullable;
 
 /**
  * Implementation of the {@link LogRecordProcessor} that batches logs exported by the SDK then
@@ -159,6 +165,9 @@ private static final class Worker implements Runnable {
     private volatile boolean continueWork = true;
     private final ArrayList<LogRecordData> batch;
     private final long maxQueueSize;
+    @Nullable private final ExecutorService exportExecutor;
+    private final Set<CompletableResultCode> pendingExports =
+        Collections.newSetFromMap(new ConcurrentHashMap<CompletableResultCode, Boolean>());
 
     private Worker(
         LogRecordExporter logRecordExporter,
@@ -180,6 +189,22 @@ private Worker(
       this.maxQueueSize = maxQueueSize;
 
       this.batch = new ArrayList<>(this.maxExportBatchSize);
+      int concurrencyLimit = logRecordExporter.getConcurrencyLimit();
+      if (concurrencyLimit < 1) {
+        throw new IllegalArgumentException("log exporter concurrency limit must be positive");
+      }
+      exportExecutor =
+          concurrencyLimit == 1
+              ? null
+              : new ThreadPoolExecutor(
+                  concurrencyLimit - 1,
+                  concurrencyLimit - 1,
+                  0,
+                  TimeUnit.NANOSECONDS,
+                  new SynchronousQueue<>(),
+                  new DaemonThreadFactory(
+                      BatchLogRecordProcessor.class.getSimpleName() + "_Exporter"),
+                  new ThreadPoolExecutor.CallerRunsPolicy());
     }
 
     private void addLog(ReadWriteLogRecord logData) {
@@ -240,6 +265,7 @@ private void flush() {
         }
       }
       exportCurrentBatch();
+      waitForExports();
       CompletableResultCode flushResult = flushRequested.get();
       if (flushResult != null) {
         flushResult.succeed();
@@ -264,6 +290,9 @@ private CompletableResultCode shutdown() {
             CompletableResultCode shutdownResult = logRecordExporter.shutdown();
             shutdownResult.whenComplete(
                 () -> {
+                  if (exportExecutor != null) {
+                    exportExecutor.shutdownNow();
+                  }
                   if (!flushResult.isSuccess() || !shutdownResult.isSuccess()) {
                     result.fail();
                   } else {
@@ -293,21 +322,61 @@ private void exportCurrentBatch() {
         return;
       }
 
+      if (exportExecutor != null) {
+        List<LogRecordData> exportBatch = new ArrayList<>(batch);
+        batch.clear();
+        CompletableResultCode taskResult = new CompletableResultCode();
+        pendingExports.add(taskResult);
+        try {
+          exportExecutor.execute(
+              () -> {
+                try {
+                  exportBatch(exportBatch);
+                  taskResult.succeed();
+                } catch (Throwable t) {
+                  ThrowableUtil.propagateIfFatal(t);
+                  taskResult.fail();
+                } finally {
+                  pendingExports.remove(taskResult);
+                }
+              });
+        } catch (Throwable t) {
+          pendingExports.remove(taskResult);
+          taskResult.fail();
+          ThrowableUtil.propagateIfFatal(t);
+          logger.log(Level.WARNING, "Unable to submit export", t);
+        }
+        return;
+      }
+
+      try {
+        exportBatch(batch);
+      } finally {
+        batch.clear();
+      }
+    }
+
+    private void exportBatch(List<LogRecordData> exportBatch) {
       try {
         // We always increment for every export invocation, so we increment before the export call
         // to make sure thrown errors don't affect it.
-        logProcessorInstrumentation.finishLogs(batch.size());
+        logProcessorInstrumentation.finishLogs(exportBatch.size());
         CompletableResultCode result =
-            logRecordExporter.export(Collections.unmodifiableList(batch));
+            logRecordExporter.export(Collections.unmodifiableList(exportBatch));
         result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
         if (!result.isSuccess()) {
           logger.log(Level.FINE, "Exporter failed");
         }
       } catch (Throwable t) {
         ThrowableUtil.propagateIfFatal(t);
         logger.log(Level.WARNING, "Exporter threw an Exception", t);
-      } finally {
-        batch.clear();
+      }
+    }
+
+    private void waitForExports() {
+      while (!pendingExports.isEmpty()) {
+        CompletableResultCode.ofAll(new ArrayList<>(pendingExports))
+            .join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
       }
     }
   }

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/LogRecordExporter.java

@@ -73,6 +73,20 @@ static LogRecordExporter noop() {
    */
   CompletableResultCode export(Collection<LogRecordData> logs);
 
+  /**
+   * Returns the maximum number of export calls that this exporter can process concurrently.
+   *
+   * <p>The default value preserves the historical serialized behavior. Exporters that support
+   * concurrent calls may return a larger value and may block an export call when this capacity is
+   * exhausted to apply backpressure.
+   *
+   * @return the export concurrency limit, which must be positive
+   * @since 1.67.0
+   */
+  default int getConcurrencyLimit() {
+    return 1;
+  }
+
   /**
    * Exports the collection of {@link LogRecordData} that have not yet been exported.
    *

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderMetricsTest.java

@@ -146,6 +146,7 @@ void batch() throws Exception {
     MeterProvider meterProvider =
         SdkMeterProvider.builder().registerMetricReader(metricReader).build();
 
+    when(mockExporter.getConcurrencyLimit()).thenReturn(1);
     BatchLogRecordProcessor processor =
         BatchLogRecordProcessor.builder(mockExporter)
             .setMaxQueueSize(1)

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java

@@ -57,6 +57,7 @@ class BatchLogRecordProcessorTest {
   @BeforeEach
   void setUp() {
     when(mockLogRecordExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
+    when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(1);
   }
 
   private void emitLog(SdkLoggerProvider sdkLoggerProvider, String message) {
@@ -439,6 +440,7 @@ public void continuesIfExporterTimesOut() throws InterruptedException {
     CountDownLatch exportedAgain = new CountDownLatch(1);
     reset(mockLogRecordExporter);
     when(mockLogRecordExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
+    when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(1);
     when(mockLogRecordExporter.export(
             argThat(
                 logs -> {
@@ -497,6 +499,46 @@ void shutdownPropagatesFailure() {
     assertThat(result.isSuccess()).isFalse();
   }
 
+  @Test
+  @Timeout(5)
+  void exporterConcurrencyLimitAllowsConcurrentExports() throws InterruptedException {
+    ConcurrentLogRecordExporter exporter = new ConcurrentLogRecordExporter();
+    SdkLoggerProvider loggerProvider =
+        SdkLoggerProvider.builder()
+            .addLogRecordProcessor(
+                BatchLogRecordProcessor.builder(exporter)
+                    .setMaxExportBatchSize(1)
+                    .setScheduleDelay(1, TimeUnit.HOURS)
+                    .build())
+            .build();
+
+    emitLog(loggerProvider, LOG_MESSAGE_1);
+    emitLog(loggerProvider, LOG_MESSAGE_2);
+    CompletableResultCode shutdown = loggerProvider.shutdown();
+
+    assertThat(exporter.entered.await(1, TimeUnit.SECONDS)).isTrue();
+    assertThat(exporter.maxConcurrent.get()).isEqualTo(2);
+    exporter.release.countDown();
+    shutdown.join(1, TimeUnit.SECONDS);
+    assertThat(shutdown.isSuccess()).isTrue();
+  }
+
+  @Test
+  void rejectsNegativeExporterConcurrencyLimit() {
+    when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(-1);
+    assertThatThrownBy(() -> BatchLogRecordProcessor.builder(mockLogRecordExporter).build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("log exporter concurrency limit must be positive");
+  }
+
+  @Test
+  void rejectsZeroExporterConcurrencyLimit() {
+    when(mockLogRecordExporter.getConcurrencyLimit()).thenReturn(0);
+    assertThatThrownBy(() -> BatchLogRecordProcessor.builder(mockLogRecordExporter).build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("log exporter concurrency limit must be positive");
+  }
+
   @Test
   void getLogRecordExporter() {
     assertThat(
@@ -578,6 +620,43 @@ private void unblock() {
     }
   }
 
+  private static final class ConcurrentLogRecordExporter implements LogRecordExporter {
+    private final CountDownLatch entered = new CountDownLatch(2);
+    private final CountDownLatch release = new CountDownLatch(1);
+    private final AtomicInteger active = new AtomicInteger();
+    private final AtomicInteger maxConcurrent = new AtomicInteger();
+
+    @Override
+    public int getConcurrencyLimit() {
+      return 2;
+    }
+
+    @Override
+    public CompletableResultCode export(Collection<LogRecordData> logs) {
+      int current = active.incrementAndGet();
+      maxConcurrent.accumulateAndGet(current, Math::max);
+      entered.countDown();
+      try {
+        release.await();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      } finally {
+        active.decrementAndGet();
+      }
+      return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+      return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+      return CompletableResultCode.ofSuccess();
+    }
+  }
+
   private static class CompletableLogRecordExporter implements LogRecordExporter {
 
     private final List<CompletableResultCode> results = new ArrayList<>();

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

@@ -60,6 +60,20 @@ default MemoryMode getMemoryMode() {
    */
   CompletableResultCode export(Collection<MetricData> metrics);
 
+  /**
+   * Returns the maximum number of export calls that this exporter can process concurrently.
+   *
+   * <p>The default value preserves the historical serialized behavior. Exporters that support
+   * concurrent calls may return a larger value and may block an export call when this capacity is
+   * exhausted to apply backpressure.
+   *
+   * @return the export concurrency limit, which must be positive
+   * @since 1.67.0
+   */
+  default int getConcurrencyLimit() {
+    return 1;
+  }
+
   /**
    * A hint that any metrics previously {@link #export(Collection)}ed should be completed.
    *