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

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Performance
+
+- Reduce the number of SDK threads: the log and metrics batch processors now share a single executor instead of each creating its own ([#5818](https://github.com/getsentry/sentry-java/pull/5818))
+
 ## 8.50.0
 
 ### Android 17 support

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 }