feat: Add async event processing

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 
+- Add opt-in async event processing ([#5558](https://github.com/getsentry/sentry-java/pull/5558))
 - Add `enableStandaloneAppStartTracing` option to send app start as a standalone transaction instead of attaching it as a child span of the first activity transaction ([#5342](https://github.com/getsentry/sentry-java/pull/5342))
   - Disabled by default; opt in via `options.isEnableStandaloneAppStartTracing = true` or manifest meta-data `io.sentry.standalone-app-start-tracing.enable`
   - Emits a transaction named `App Start` with op `app.start`, carrying the existing app start measurements and phase spans (`process.load`, `contentprovider.load`, `application.load`, activity lifecycle spans) as direct children of the root

docs/superpowers/specs/2026-06-15-async-event-processing-design.md

@@ -0,0 +1,126 @@
+# Async Event Processing Design
+
+## Context
+
+The Java SDK currently invokes `EventProcessor.process(...)` synchronously in `SentryClient` before the relevant `beforeSend*` callback and before enqueueing work for transport or batching. This means custom processor and `beforeSend*` code can run on the caller thread.
+
+We want to introduce an opt-in mode that moves the late mutation/drop phase off the caller thread while keeping the existing synchronous processor phase in place. The first iteration should avoid Android public API compatibility risk from `CompletionStage`/`CompletableFuture`, so async processor methods use the same return shape as the existing synchronous methods.
+
+## Goals
+
+- Add async event processor callbacks for every event type currently supported by `EventProcessor`.
+- Keep existing synchronous `process(...)` callbacks in their current positions.
+- Add an opt-in `SentryOptions.enableAsyncProcessing` option, defaulting to `false`.
+- When async processing is enabled, return from `capture*` after the item is accepted by the async processing queue.
+- Isolate potentially slow user callbacks from the shared `SentryOptions.executorService`.
+- Make `flush` and `close` wait for accepted async processing work.
+
+## Non-goals
+
+- Do not add `CompletionStage`, `CompletableFuture`, Kotlin coroutines, or other asynchronous return types to the public API.
+- Do not change session, check-in, or profile chunk processing; these paths do not use `EventProcessor` today.
+- Do not remove or move existing synchronous `process(...)` callbacks.
+- Do not guarantee that a returned event ID means the item reaches Sentry.
+
+## Public API
+
+`EventProcessor` gets default `processAsync(...)` overloads that mirror the current `process(...)` overloads:
+
+- `SentryEvent processAsync(SentryEvent event, Hint hint)`
+- `SentryTransaction processAsync(SentryTransaction transaction, Hint hint)`
+- `SentryReplayEvent processAsync(SentryReplayEvent event, Hint hint)`
+- `SentryLogEvent processAsync(SentryLogEvent event)`
+- `SentryMetricsEvent processAsync(SentryMetricsEvent event, Hint hint)`
+
+Each default implementation returns the input item unchanged. Returning `null` drops the item, matching existing processor semantics.
+
+`SentryOptions` gets a direct boolean option:
+
+- `isEnableAsyncProcessing()`
+- `setEnableAsyncProcessing(boolean enableAsyncProcessing)`
+
+The default is `false`. External configuration support should use the key `enable-async-processing`.
+
+## Runtime Architecture
+
+`SentryClient` owns a dedicated bounded single-thread async processing queue. The queue is separate from `SentryOptions.executorService` so user processor and `beforeSend*` code cannot clog SDK maintenance tasks.
+
+The async processing queue is active only when `enableAsyncProcessing=true`. Its capacity reuses `SentryOptions.maxQueueSize` to avoid adding a second queue-size option in the first iteration.
+
+When `enableAsyncProcessing=false`, `SentryClient` still invokes the new `processAsync(...)` stage, but inline on the caller thread. This keeps the processing pipeline shape consistent and avoids conditional logic around whether async processor callbacks are invoked.
+
+## Telemetry Processor Compatibility
+
+The async processing queue is a callback execution stage, not a delivery buffer or telemetry scheduler. It owns only running `processAsync(...)`, running the matching `beforeSend*` callback, and forwarding accepted items to the existing downstream delivery layer.
+
+Future telemetry processor integration should keep batching, prioritization, overflow policy, trace-aware span bucketing, and offline/cache behavior in the telemetry processor. Async processing should happen before telemetry buffer/scheduler ingestion, and future work may replace or reshape this queue to avoid keeping both a generic FIFO callback queue and category-aware telemetry buffers.
+
+The first iteration uses one bounded FIFO queue because the feature is opt-in. This can cause priority inversion under mixed high-volume traffic, for example logs or metrics filling the callback queue before higher-priority errors reach a future scheduler. Queue overflow at this stage is recorded as `queue_overflow`; future telemetry buffers should use a distinct `buffer_overflow` reason.
+
+`flush(timeoutMillis)` and `close()` must treat async processing and downstream queues as one pipeline and use a shared deadline instead of giving every layer the full timeout independently.
+
+## Processing Pipeline
+
+For each supported capture path, the order is:
+
+1. Existing early filtering and scope work.
+2. Existing scoped synchronous `process(...)`, where applicable.
+3. Existing global synchronous `process(...)`.
+4. New scoped `processAsync(...)`, where applicable.
+5. New global `processAsync(...)`.
+6. Matching `beforeSend*` callback.
+7. Existing send, log batch, or metrics batch queue.
+
+Supported paths:
+
+- Error events: async processors run before `beforeSend`.
+- Feedback events: async processors run before `beforeSendFeedback`.
+- Transactions: async processors run before `beforeSendTransaction`.
+- Replay events: async processors run before `beforeSendReplay`.
+- Logs: async processors run before logs `beforeSend`.
+- Metrics: async processors run before metrics `beforeSend`.
+
+Sessions, check-ins, and profile chunks are unchanged.
+
+## Capture Return Behavior
+
+When `enableAsyncProcessing=true`, ID-returning capture methods enqueue the late processing task and return immediately if enqueue succeeds. They return the event, transaction, replay, or feedback ID even though the item may later be dropped by async processors, `beforeSend*`, sampling, rate limiting, transport failures, or downstream queue overflow.
+
+If the async processing queue is full, enqueue fails immediately. The SDK drops the item, records `DiscardReason.QUEUE_OVERFLOW`, and returns `SentryId.EMPTY_ID` for ID-returning capture methods.
+
+Log and metrics capture methods do not return an ID. If the async processing queue is full, they drop the item and record the appropriate queue-overflow client report.
+
+## Error Handling and Client Reports
+
+`processAsync(...)` uses the same error policy as `process(...)`:
+
+- Exceptions from processors are logged.
+- Processing continues with the current item.
+- Returning `null` drops the item.
+- Drops are recorded with `DiscardReason.EVENT_PROCESSOR` and the existing data category for the item type.
+
+`beforeSend*` keeps its current behavior. Exceptions drop the item for PII safety and record `DiscardReason.BEFORE_SEND` where applicable.
+
+Transaction accounting must remain consistent:
+
+- Dropping a transaction records the transaction and all spans.
+- Removing spans from a transaction in an async processor or `beforeSendTransaction` records the dropped span count.
+
+## Flush and Close
+
+`SentryClient.flush(timeoutMillis)` waits for the async processing queue to become idle and then waits for downstream log, metrics, and transport queues.
+
+`SentryClient.close()` drains async processing during shutdown and then closes downstream processors and transport. Closing remains bounded by existing shutdown and flush timeout behavior.
+
+## Testing Plan
+
+- Verify `EventProcessor.processAsync(...)` default methods return input items unchanged.
+- Verify `SentryOptions.enableAsyncProcessing` defaults to `false` and can be set directly and through external options.
+- Verify inline mode invokes `processAsync(...)` on the caller thread before the matching `beforeSend*` callback.
+- Verify async-enabled mode returns before `processAsync(...)` and `beforeSend*` finish.
+- Verify ordering: scoped sync processor, global sync processor, scoped async processor, global async processor, `beforeSend*`.
+- Verify async queue overflow drops the item, records `queue_overflow`, and returns `SentryId.EMPTY_ID` where applicable.
+- Verify `flush` waits for accepted async processing work.
+- Verify processor exceptions are logged and do not drop the item.
+- Verify `processAsync(...)` returning `null` records `event_processor` drops for events, feedback, transactions and spans, replay, logs, and metrics.
+- Verify `beforeSendTransaction` and async transaction processors preserve existing span-loss accounting.

sentry/api/sentry.api

@@ -474,6 +474,11 @@ public abstract interface class io/sentry/EventProcessor {
 	public fun process (Lio/sentry/SentryMetricsEvent;Lio/sentry/Hint;)Lio/sentry/SentryMetricsEvent;
 	public fun process (Lio/sentry/SentryReplayEvent;Lio/sentry/Hint;)Lio/sentry/SentryReplayEvent;
 	public fun process (Lio/sentry/protocol/SentryTransaction;Lio/sentry/Hint;)Lio/sentry/protocol/SentryTransaction;
+	public fun processAsync (Lio/sentry/SentryEvent;Lio/sentry/Hint;)Lio/sentry/SentryEvent;
+	public fun processAsync (Lio/sentry/SentryLogEvent;)Lio/sentry/SentryLogEvent;
+	public fun processAsync (Lio/sentry/SentryMetricsEvent;Lio/sentry/Hint;)Lio/sentry/SentryMetricsEvent;
+	public fun processAsync (Lio/sentry/SentryReplayEvent;Lio/sentry/Hint;)Lio/sentry/SentryReplayEvent;
+	public fun processAsync (Lio/sentry/protocol/SentryTransaction;Lio/sentry/Hint;)Lio/sentry/protocol/SentryTransaction;
 }
 
 public final class io/sentry/ExperimentalOptions {
@@ -525,6 +530,7 @@ public final class io/sentry/ExternalOptions {
 	public fun getTracePropagationTargets ()Ljava/util/List;
 	public fun getTracesSampleRate ()Ljava/lang/Double;
 	public fun isCaptureOpenTelemetryEvents ()Ljava/lang/Boolean;
+	public fun isEnableAsyncProcessing ()Ljava/lang/Boolean;
 	public fun isEnableBackpressureHandling ()Ljava/lang/Boolean;
 	public fun isEnableCacheTracing ()Ljava/lang/Boolean;
 	public fun isEnableDatabaseTransactionTracing ()Ljava/lang/Boolean;
@@ -544,6 +550,7 @@ public final class io/sentry/ExternalOptions {
 	public fun setDebug (Ljava/lang/Boolean;)V
 	public fun setDist (Ljava/lang/String;)V
 	public fun setDsn (Ljava/lang/String;)V
+	public fun setEnableAsyncProcessing (Ljava/lang/Boolean;)V
 	public fun setEnableBackpressureHandling (Ljava/lang/Boolean;)V
 	public fun setEnableCacheTracing (Ljava/lang/Boolean;)V
 	public fun setEnableDatabaseTransactionTracing (Ljava/lang/Boolean;)V
@@ -3717,6 +3724,7 @@ public class io/sentry/SentryOptions {
 	public fun isContinuousProfilingEnabled ()Z
 	public fun isDebug ()Z
 	public fun isEnableAppStartProfiling ()Z
+	public fun isEnableAsyncProcessing ()Z
 	public fun isEnableAutoSessionTracking ()Z
 	public fun isEnableBackpressureHandling ()Z
 	public fun isEnableCacheTracing ()Z
@@ -3778,6 +3786,7 @@ public class io/sentry/SentryOptions {
 	public fun setDistributionController (Lio/sentry/IDistributionApi;)V
 	public fun setDsn (Ljava/lang/String;)V
 	public fun setEnableAppStartProfiling (Z)V
+	public fun setEnableAsyncProcessing (Z)V
 	public fun setEnableAutoSessionTracking (Z)V
 	public fun setEnableBackpressureHandling (Z)V
 	public fun setEnableCacheTracing (Z)V

sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java

@@ -0,0 +1,87 @@
+package io.sentry;
+
+import io.sentry.transport.ReusableCountLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.jetbrains.annotations.ApiStatus;
+import org.jetbrains.annotations.NotNull;
+
+@ApiStatus.Internal
+final class AsyncEventProcessingExecutor {
+  private final int maxQueueSize;
+  private final @NotNull ILogger logger;
+  private final @NotNull ThreadPoolExecutor executor;
+  private final @NotNull ReusableCountLatch unfinishedTasksCount = new ReusableCountLatch();
+  private boolean closed = false;
+
+  AsyncEventProcessingExecutor(final int maxQueueSize, final @NotNull ILogger logger) {
+    this.maxQueueSize = maxQueueSize;
+    this.logger = logger;
+    this.executor =
+        new ThreadPoolExecutor(
+            1,
+            1,
+            0L,
+            TimeUnit.MILLISECONDS,
+            new LinkedBlockingQueue<>(),
+            new AsyncEventProcessingThreadFactory());
+  }
+
+  synchronized boolean submit(final @NotNull Runnable task) {
+    if (closed || unfinishedTasksCount.getCount() >= maxQueueSize) {
+      return false;
+    }
+
+    unfinishedTasksCount.increment();
+    try {
+      executor.execute(
+          () -> {
+            try {
+              task.run();
+            } finally {
+              unfinishedTasksCount.decrement();
+            }
+          });
+      return true;
+    } catch (Throwable e) {
+      unfinishedTasksCount.decrement();
+      logger.log(SentryLevel.WARNING, "Async event processing task rejected.", e);
+      return false;
+    }
+  }
+
+  void waitTillIdle(final long timeoutMillis) {
+    try {
+      unfinishedTasksCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      logger.log(SentryLevel.ERROR, "Failed to wait for async event processing queue to drain.", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  synchronized void close(final long timeoutMillis) {
+    closed = true;
+    executor.shutdown();
+    try {
+      if (!executor.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS)) {
+        executor.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      executor.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private static final class AsyncEventProcessingThreadFactory implements ThreadFactory {
+    private int cnt;
+
+    @Override
+    public @NotNull Thread newThread(final @NotNull Runnable r) {
+      final Thread thread = new Thread(r, "SentryAsyncEventProcessing-" + cnt++);
+      thread.setDaemon(true);
+      return thread;
+    }
+  }
+}

sentry/src/main/java/io/sentry/EventProcessor.java

@@ -67,6 +67,66 @@ default SentryMetricsEvent process(@NotNull SentryMetricsEvent event, @NotNull H
     return event;
   }
 
+  /**
+   * May mutate or drop a SentryEvent during the async processing stage.
+   *
+   * @param event the SentryEvent
+   * @param hint the Hint
+   * @return the event itself, a mutated SentryEvent or null
+   */
+  @Nullable
+  default SentryEvent processAsync(@NotNull SentryEvent event, @NotNull Hint hint) {
+    return event;
+  }
+
+  /**
+   * May mutate or drop a SentryTransaction during the async processing stage.
+   *
+   * @param transaction the SentryTransaction
+   * @param hint the Hint
+   * @return the event itself, a mutated SentryTransaction or null
+   */
+  @Nullable
+  default SentryTransaction processAsync(
+      @NotNull SentryTransaction transaction, @NotNull Hint hint) {
+    return transaction;
+  }
+
+  /**
+   * May mutate or drop a SentryReplayEvent during the async processing stage.
+   *
+   * @param event the SentryReplayEvent
+   * @param hint the Hint
+   * @return the event itself, a mutated SentryReplayEvent or null
+   */
+  @Nullable
+  default SentryReplayEvent processAsync(@NotNull SentryReplayEvent event, @NotNull Hint hint) {
+    return event;
+  }
+
+  /**
+   * May mutate or drop a SentryLogEvent during the async processing stage.
+   *
+   * @param event the SentryLogEvent
+   * @return the event itself, a mutated SentryLogEvent or null
+   */
+  @Nullable
+  default SentryLogEvent processAsync(@NotNull SentryLogEvent event) {
+    return event;
+  }
+
+  /**
+   * May mutate or drop a SentryMetricsEvent during the async processing stage.
+   *
+   * @param event the SentryMetricsEvent
+   * @param hint the Hint
+   * @return the event itself, a mutated SentryMetricsEvent or null
+   */
+  @Nullable
+  default SentryMetricsEvent processAsync(@NotNull SentryMetricsEvent event, @NotNull Hint hint) {
+    return event;
+  }
+
   /**
    * Controls when this EventProcessor is invoked.
    *

sentry/src/main/java/io/sentry/ExternalOptions.java

@@ -23,6 +23,7 @@ public final class ExternalOptions {
   private @Nullable Boolean enableUncaughtExceptionHandler;
   private @Nullable Boolean debug;
   private @Nullable Boolean enableDeduplication;
+  private @Nullable Boolean enableAsyncProcessing;
   private @Nullable Double sampleRate;
   private @Nullable Double tracesSampleRate;
   private @Nullable Double profilesSampleRate;
@@ -90,6 +91,8 @@ public final class ExternalOptions {
     options.setProfilesSampleRate(propertiesProvider.getDoubleProperty("profiles-sample-rate"));
     options.setDebug(propertiesProvider.getBooleanProperty("debug"));
     options.setEnableDeduplication(propertiesProvider.getBooleanProperty("enable-deduplication"));
+    options.setEnableAsyncProcessing(
+        propertiesProvider.getBooleanProperty("enable-async-processing"));
     options.setSendClientReports(propertiesProvider.getBooleanProperty("send-client-reports"));
     options.setForceInit(propertiesProvider.getBooleanProperty("force-init"));
     final String maxRequestBodySize = propertiesProvider.getProperty("max-request-body-size");
@@ -315,6 +318,14 @@ public void setEnableDeduplication(final @Nullable Boolean enableDeduplication)
     this.enableDeduplication = enableDeduplication;
   }
 
+  public @Nullable Boolean isEnableAsyncProcessing() {
+    return enableAsyncProcessing;
+  }
+
+  public void setEnableAsyncProcessing(final @Nullable Boolean enableAsyncProcessing) {
+    this.enableAsyncProcessing = enableAsyncProcessing;
+  }
+
   public @Nullable Double getSampleRate() {
     return sampleRate;
   }