Add support for setting OTLP exporter executor service (#7152)

Author: jack-berg

docs/apidiffs/current_vs_latest/opentelemetry-exporter-otlp.txt

@@ -1,2 +1,19 @@
 Comparing source compatibility of opentelemetry-exporter-otlp-1.49.0-SNAPSHOT.jar against opentelemetry-exporter-otlp-1.48.0.jar
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder setExecutorService(java.util.concurrent.ExecutorService)

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterBuilder.java

@@ -25,6 +25,7 @@
 import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.StringJoiner;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
@@ -63,6 +64,7 @@ public class GrpcExporterBuilder<T extends Marshaler> {
   @Nullable private RetryPolicy retryPolicy = RetryPolicy.getDefault();
   private Supplier<MeterProvider> meterProviderSupplier = GlobalOpenTelemetry::getMeterProvider;
   private ClassLoader serviceClassLoader = GrpcExporterBuilder.class.getClassLoader();
+  @Nullable private ExecutorService executorService;
 
   // Use Object type since gRPC may not be on the classpath.
   @Nullable private Object grpcChannel;
@@ -153,6 +155,11 @@ public GrpcExporterBuilder<T> setServiceClassLoader(ClassLoader servieClassLoade
     return this;
   }
 
+  public GrpcExporterBuilder<T> setExecutorService(ExecutorService executorService) {
+    this.executorService = executorService;
+    return this;
+  }
+
   @SuppressWarnings("BuilderReturnThis")
   public GrpcExporterBuilder<T> copy() {
     GrpcExporterBuilder<T> copy =
@@ -216,7 +223,8 @@ public GrpcExporter<T> build() {
                 grpcStubFactory,
                 retryPolicy,
                 isPlainHttp ? null : tlsConfigHelper.getSslContext(),
-                isPlainHttp ? null : tlsConfigHelper.getTrustManager()));
+                isPlainHttp ? null : tlsConfigHelper.getTrustManager(),
+                executorService));
     LOGGER.log(Level.FINE, "Using GrpcSender: " + grpcSender.getClass().getName());
 
     return new GrpcExporter<>(exporterName, type, grpcSender, meterProviderSupplier);
@@ -250,6 +258,9 @@ public String toString(boolean includePrefixAndSuffix) {
       joiner.add("grpcChannel=" + grpcChannel);
     }
     joiner.add("serviceClassLoader=" + serviceClassLoader);
+    if (executorService != null) {
+      joiner.add("executorService=" + executorService);
+    }
     // Note: omit tlsConfigHelper because we can't log the configuration in any readable way
     // Note: omit meterProviderSupplier because we can't log the configuration in any readable way
     return joiner.toString();

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcSenderConfig.java

@@ -13,6 +13,7 @@
 import java.net.URI;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -40,7 +41,8 @@ public static <T extends Marshaler> GrpcSenderConfig<T> create(
       Supplier<BiFunction<Channel, String, MarshalerServiceStub<T, ?, ?>>> stubFactory,
       @Nullable RetryPolicy retryPolicy,
       @Nullable SSLContext sslContext,
-      @Nullable X509TrustManager trustManager) {
+      @Nullable X509TrustManager trustManager,
+      @Nullable ExecutorService executorService) {
     return new AutoValue_GrpcSenderConfig<>(
         endpoint,
         endpointPath,
@@ -52,7 +54,8 @@ public static <T extends Marshaler> GrpcSenderConfig<T> create(
         stubFactory,
         retryPolicy,
         sslContext,
-        trustManager);
+        trustManager,
+        executorService);
   }
 
   public abstract URI getEndpoint();
@@ -82,4 +85,7 @@ public static <T extends Marshaler> GrpcSenderConfig<T> create(
 
   @Nullable
   public abstract X509TrustManager getTrustManager();
+
+  @Nullable
+  public abstract ExecutorService getExecutorService();
 }

exporters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpExporterBuilder.java

@@ -23,6 +23,7 @@
 import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.StringJoiner;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.logging.Level;
@@ -61,6 +62,7 @@ public final class HttpExporterBuilder<T extends Marshaler> {
   @Nullable private RetryPolicy retryPolicy = RetryPolicy.getDefault();
   private Supplier<MeterProvider> meterProviderSupplier = GlobalOpenTelemetry::getMeterProvider;
   private ClassLoader serviceClassLoader = HttpExporterBuilder.class.getClassLoader();
+  @Nullable private ExecutorService executorService;
 
   public HttpExporterBuilder(String exporterName, String type, String defaultEndpoint) {
     this.exporterName = exporterName;
@@ -137,6 +139,11 @@ public HttpExporterBuilder<T> setServiceClassLoader(ClassLoader servieClassLoade
     return this;
   }
 
+  public HttpExporterBuilder<T> setExecutorService(ExecutorService executorService) {
+    this.executorService = executorService;
+    return this;
+  }
+
   public HttpExporterBuilder<T> exportAsJson() {
     this.exportAsJson = true;
     return this;
@@ -198,7 +205,8 @@ public HttpExporter<T> build() {
                 proxyOptions,
                 retryPolicy,
                 isPlainHttp ? null : tlsConfigHelper.getSslContext(),
-                isPlainHttp ? null : tlsConfigHelper.getTrustManager()));
+                isPlainHttp ? null : tlsConfigHelper.getTrustManager(),
+                executorService));
     LOGGER.log(Level.FINE, "Using HttpSender: " + httpSender.getClass().getName());
 
     return new HttpExporter<>(exporterName, type, httpSender, meterProviderSupplier, exportAsJson);
@@ -230,6 +238,9 @@ public String toString(boolean includePrefixAndSuffix) {
       joiner.add("retryPolicy=" + retryPolicy);
     }
     joiner.add("serviceClassLoader=" + serviceClassLoader);
+    if (executorService != null) {
+      joiner.add("executorService=" + executorService);
+    }
     // Note: omit tlsConfigHelper because we can't log the configuration in any readable way
     // Note: omit meterProviderSupplier because we can't log the configuration in any readable way
     return joiner.toString();

exporters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpSenderConfig.java

@@ -11,6 +11,7 @@
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
@@ -37,7 +38,8 @@ public static HttpSenderConfig create(
       @Nullable ProxyOptions proxyOptions,
       @Nullable RetryPolicy retryPolicy,
       @Nullable SSLContext sslContext,
-      @Nullable X509TrustManager trustManager) {
+      @Nullable X509TrustManager trustManager,
+      @Nullable ExecutorService executorService) {
     return new AutoValue_HttpSenderConfig(
         endpoint,
         compressor,
@@ -49,7 +51,8 @@ public static HttpSenderConfig create(
         proxyOptions,
         retryPolicy,
         sslContext,
-        trustManager);
+        trustManager,
+        executorService);
   }
 
   public abstract String getEndpoint();
@@ -78,4 +81,7 @@ public static HttpSenderConfig create(
 
   @Nullable
   public abstract X509TrustManager getTrustManager();
+
+  @Nullable
+  public abstract ExecutorService getExecutorService();
 }

exporters/otlp/all/src/jmh/java/io/opentelemetry/exporter/otlp/trace/OltpExporterBenchmark.java

@@ -88,7 +88,8 @@ public void setUp() {
                 MarshalerTraceServiceGrpc.newFutureStub(defaultGrpcChannel, null),
                 /* shutdownChannel= */ false,
                 10,
-                Collections::emptyMap),
+                Collections::emptyMap,
+                null),
             MeterProvider::noop);
 
     okhttpGrpcSender =
@@ -105,6 +106,7 @@ public void setUp() {
                 Collections::emptyMap,
                 null,
                 null,
+                null,
                 null),
             MeterProvider::noop);
 

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/logs/OtlpHttpLogRecordExporterBuilder.java

@@ -21,6 +21,7 @@
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.time.Duration;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -236,6 +237,19 @@ public OtlpHttpLogRecordExporterBuilder setServiceClassLoader(ClassLoader servic
     return this;
   }
 
+  /**
+   * Set the {@link ExecutorService} used to execute requests.
+   *
+   * <p>NOTE: By calling this method, you are opting into managing the lifecycle of the {@code
+   * executorService}. {@link ExecutorService#shutdown()} will NOT be called when this exporter is
+   * shutdown.
+   */
+  public OtlpHttpLogRecordExporterBuilder setExecutorService(ExecutorService executorService) {
+    requireNonNull(executorService, "executorService");
+    delegate.setExecutorService(executorService);
+    return this;
+  }
+
   /**
    * Constructs a new instance of the exporter based on the builder's values.
    *

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilder.java

@@ -26,6 +26,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -264,6 +265,19 @@ public OtlpHttpMetricExporterBuilder setServiceClassLoader(ClassLoader serviceCl
     return this;
   }
 
+  /**
+   * Set the {@link ExecutorService} used to execute requests.
+   *
+   * <p>NOTE: By calling this method, you are opting into managing the lifecycle of the {@code
+   * executorService}. {@link ExecutorService#shutdown()} will NOT be called when this exporter is
+   * shutdown.
+   */
+  public OtlpHttpMetricExporterBuilder setExecutorService(ExecutorService executorService) {
+    requireNonNull(executorService, "executorService");
+    delegate.setExecutorService(executorService);
+    return this;
+  }
+
   OtlpHttpMetricExporterBuilder exportAsJson() {
     delegate.exportAsJson();
     return this;

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/trace/OtlpHttpSpanExporterBuilder.java

@@ -21,6 +21,7 @@
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.time.Duration;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -237,6 +238,19 @@ public OtlpHttpSpanExporterBuilder setServiceClassLoader(ClassLoader serviceClas
     return this;
   }
 
+  /**
+   * Set the {@link ExecutorService} used to execute requests.
+   *
+   * <p>NOTE: By calling this method, you are opting into managing the lifecycle of the {@code
+   * executorService}. {@link ExecutorService#shutdown()} will NOT be called when this exporter is
+   * shutdown.
+   */
+  public OtlpHttpSpanExporterBuilder setExecutorService(ExecutorService executorService) {
+    requireNonNull(executorService, "executorService");
+    delegate.setExecutorService(executorService);
+    return this;
+  }
+
   /**
    * Constructs a new instance of the exporter based on the builder's values.
    *

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/logs/OtlpGrpcLogRecordExporterBuilder.java

@@ -22,6 +22,7 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -269,6 +270,19 @@ public OtlpGrpcLogRecordExporterBuilder setServiceClassLoader(ClassLoader servic
     return this;
   }
 
+  /**
+   * Set the {@link ExecutorService} used to execute requests.
+   *
+   * <p>NOTE: By calling this method, you are opting into managing the lifecycle of the {@code
+   * executorService}. {@link ExecutorService#shutdown()} will NOT be called when this exporter is
+   * shutdown.
+   */
+  public OtlpGrpcLogRecordExporterBuilder setExecutorService(ExecutorService executorService) {
+    requireNonNull(executorService, "executorService");
+    delegate.setExecutorService(executorService);
+    return this;
+  }
+
   /**
    * Constructs a new instance of the exporter based on the builder's values.
    *