perf: Share one executor between log and metrics batch processors (JAVA-653)

The log and metrics batch processors each created their own
SentryExecutorService, so an app using both spawned two threads for
work that is identical in shape (a 5s flush loop that hands envelopes
to the transport). SentryClient now owns a single executor, created
only when logs or metrics are enabled, and injects it into both
processors via the default factories. The restart shutdown path
tolerates the shared executor already being closed by the sibling
processor.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: runningcode

sentry/api/sentry.api

@@ -3018,6 +3018,7 @@ public final class io/sentry/SentryClient : io/sentry/ISentryClient {
 	public fun close ()V
 	public fun close (Z)V
 	public fun flush (J)V
+	public fun getBatchProcessorExecutorService ()Lio/sentry/ISentryExecutorService;
 	public fun getRateLimiter ()Lio/sentry/transport/RateLimiter;
 	public fun isEnabled ()Z
 	public fun isHealthy ()Z
@@ -5475,6 +5476,7 @@ public class io/sentry/metrics/MetricsBatchProcessor : io/sentry/metrics/IMetric
 	public static final field MAX_QUEUE_SIZE I
 	protected final field options Lio/sentry/SentryOptions;
 	public fun <init> (Lio/sentry/SentryOptions;Lio/sentry/ISentryClient;)V
+	public fun <init> (Lio/sentry/SentryOptions;Lio/sentry/ISentryClient;Lio/sentry/ISentryExecutorService;)V
 	public fun add (Lio/sentry/SentryMetricsEvent;)V
 	public fun close (Z)V
 	public fun flush (J)V

sentry/src/main/java/io/sentry/SentryClient.java

@@ -45,6 +45,10 @@ public final class SentryClient implements ISentryClient {
   private final @NotNull ILoggerBatchProcessor loggerBatchProcessor;
   private final @NotNull IMetricsBatchProcessor metricsBatchProcessor;
 
+  // Single executor shared by the log and metrics batch processors, so they don't spawn a thread
+  // each. Created only when logs or metrics are enabled, and closed when those processors close.
+  private final @Nullable ISentryExecutorService batchProcessorExecutorService;
+
   @Override
   public boolean isEnabled() {
     return enabled;
@@ -62,6 +66,14 @@ public SentryClient(final @NotNull SentryOptions options) {
 
     final RequestDetailsResolver requestDetailsResolver = new RequestDetailsResolver(options);
     transport = transportFactory.create(options, requestDetailsResolver.resolve());
+
+    // must be set before the batch processors are created, as they read it from this client
+    if (options.getLogs().isEnabled() || options.getMetrics().isEnabled()) {
+      batchProcessorExecutorService = new SentryExecutorService(options);
+    } else {
+      batchProcessorExecutorService = null;
+    }
+
     if (options.getLogs().isEnabled()) {
       loggerBatchProcessor =
           options.getLogs().getLoggerBatchProcessorFactory().create(options, this);
@@ -76,6 +88,16 @@ public SentryClient(final @NotNull SentryOptions options) {
     }
   }
 
+  /**
+   * The executor shared by the log and metrics batch processors. Only present (non-null) when logs
+   * or metrics are enabled, which is the only time the batch processors request it.
+   */
+  @ApiStatus.Internal
+  public @NotNull ISentryExecutorService getBatchProcessorExecutorService() {
+    return Objects.requireNonNull(
+        batchProcessorExecutorService, "batch processor executor service is not available");
+  }
+
   private boolean shouldApplyScopeData(
       final @NotNull SentryBaseEvent event, final @NotNull Hint hint) {
     if (HintUtils.shouldApplyScopeData(hint)) {

sentry/src/main/java/io/sentry/logger/DefaultLoggerBatchProcessorFactory.java

@@ -8,6 +8,6 @@ public final class DefaultLoggerBatchProcessorFactory implements ILoggerBatchPro
   @Override
   public @NotNull ILoggerBatchProcessor create(
       @NotNull SentryOptions options, @NotNull SentryClient client) {
-    return new LoggerBatchProcessor(options, client);
+    return new LoggerBatchProcessor(options, client, client.getBatchProcessorExecutorService());
   }
 }

sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java

@@ -84,7 +84,13 @@ public void close(final boolean isRestarting) {
     isShuttingDown = true;
     if (isRestarting) {
       maybeSchedule(true);
-      executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
+      try {
+        executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
+      } catch (RejectedExecutionException e) {
+        // the shared executor may already be shutting down (e.g. closed by the metrics batch
+        // processor); close it directly instead
+        executorService.close(options.getShutdownTimeoutMillis());
+      }
     } else {
       executorService.close(options.getShutdownTimeoutMillis());
       while (!queue.isEmpty()) {

sentry/src/main/java/io/sentry/metrics/DefaultMetricsBatchProcessorFactory.java

@@ -8,6 +8,6 @@ public final class DefaultMetricsBatchProcessorFactory implements IMetricsBatchP
   @Override
   public @NotNull IMetricsBatchProcessor create(
       final @NotNull SentryOptions options, final @NotNull SentryClient client) {
-    return new MetricsBatchProcessor(options, client);
+    return new MetricsBatchProcessor(options, client, client.getBatchProcessorExecutorService());
   }
 }

sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java

@@ -19,8 +19,10 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.TestOnly;
 
 @Open
 public class MetricsBatchProcessor implements IMetricsBatchProcessor {
@@ -40,10 +42,19 @@ public class MetricsBatchProcessor implements IMetricsBatchProcessor {
 
   public MetricsBatchProcessor(
       final @NotNull SentryOptions options, final @NotNull ISentryClient client) {
+    this(options, client, new SentryExecutorService(options));
+  }
+
+  @ApiStatus.Internal
+  @TestOnly
+  public MetricsBatchProcessor(
+      final @NotNull SentryOptions options,
+      final @NotNull ISentryClient client,
+      final @NotNull ISentryExecutorService executorService) {
     this.options = options;
     this.client = client;
     this.queue = new ConcurrentLinkedQueue<>();
-    this.executorService = new SentryExecutorService(options);
+    this.executorService = executorService;
   }
 
   @Override
@@ -74,7 +85,13 @@ public void close(final boolean isRestarting) {
     isShuttingDown = true;
     if (isRestarting) {
       maybeSchedule(true);
-      executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
+      try {
+        executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
+      } catch (RejectedExecutionException e) {
+        // the shared executor may already be shutting down (e.g. closed by the log batch
+        // processor); close it directly instead
+        executorService.close(options.getShutdownTimeoutMillis());
+      }
     } else {
       executorService.close(options.getShutdownTimeoutMillis());
       while (!queue.isEmpty()) {

sentry/src/test/java/io/sentry/SentryClientTest.kt

@@ -29,6 +29,7 @@ import io.sentry.protocol.SentryTransaction
 import io.sentry.protocol.User
 import io.sentry.protocol.ViewHierarchy
 import io.sentry.test.callMethod
+import io.sentry.test.getProperty
 import io.sentry.test.injectForField
 import io.sentry.transport.ITransport
 import io.sentry.transport.ITransportGate
@@ -177,6 +178,33 @@ class SentryClientTest {
     assertTrue(sut.isEnabled)
   }
 
+  @Test
+  fun `log and metrics batch processors share a single executor`() {
+    // real default factories (not the fixture mocks) so the shared executor is actually wired
+    val options =
+      SentryOptions().apply {
+        dsn = dsnString
+        logs.isEnabled = true
+        metrics.isEnabled = true
+      }
+    val client = SentryClient(options)
+    try {
+      val loggerExecutor =
+        client
+          .getProperty<ILoggerBatchProcessor>("loggerBatchProcessor")
+          .getProperty<ISentryExecutorService>("executorService")
+      val metricsExecutor =
+        client
+          .getProperty<IMetricsBatchProcessor>("metricsBatchProcessor")
+          .getProperty<ISentryExecutorService>("executorService")
+
+      assertSame(client.batchProcessorExecutorService, loggerExecutor)
+      assertSame(loggerExecutor, metricsExecutor)
+    } finally {
+      client.close()
+    }
+  }
+
   @Test
   fun `when client is closed with isRestarting false, transport waits`() {
     val sut = fixture.getSut { options -> options.logs.isEnabled = true }