From c4f1a6ec87263631fd9a29c2cee36cf75b368632 Mon Sep 17 00:00:00 2001
From: Alexander Dinauer <adinauer@users.noreply.github.com>
Date: Mon, 15 Jun 2026 15:28:34 +0200
Subject: [PATCH 1/4] docs: Add async event processing design

Document the proposed opt-in async processing pipeline for EventProcessor callbacks, including queue behavior, flush semantics, and testing coverage.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 ...026-06-15-async-event-processing-design.md | 116 ++++++++++++++++++
 1 file changed, 116 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-06-15-async-event-processing-design.md

diff --git a/docs/superpowers/specs/2026-06-15-async-event-processing-design.md b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
new file mode 100644
index 00000000000..d6510dc3cc5
--- /dev/null
+++ b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
@@ -0,0 +1,116 @@
+# 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.
+
+## 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.

From d1c16b1999cefeb8921865a201c8e97456c754e3 Mon Sep 17 00:00:00 2001
From: Alexander Dinauer <adinauer@users.noreply.github.com>
Date: Tue, 16 Jun 2026 13:44:24 +0200
Subject: [PATCH 2/4] feat: Add async event processing

Add EventProcessor.processAsync callbacks and an opt-in SentryOptions flag to run late event processing off the caller thread.

Use a dedicated bounded queue for async callbacks, record queue overflow drops, and make flush/close drain accepted async work before downstream queues.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 ...026-06-15-async-event-processing-design.md |  10 +
 sentry/api/sentry.api                         |   9 +
 .../sentry/AsyncEventProcessingExecutor.java  |  87 ++++
 .../main/java/io/sentry/EventProcessor.java   |  60 +++
 .../main/java/io/sentry/ExternalOptions.java  |  11 +
 .../src/main/java/io/sentry/SentryClient.java | 460 +++++++++++++++++-
 .../main/java/io/sentry/SentryOptions.java    |  24 +
 .../java/io/sentry/ExternalOptionsTest.kt     |  12 +
 .../test/java/io/sentry/SentryClientTest.kt   | 130 ++++-
 .../test/java/io/sentry/SentryOptionsTest.kt  |  14 +
 10 files changed, 792 insertions(+), 25 deletions(-)
 create mode 100644 sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java

diff --git a/docs/superpowers/specs/2026-06-15-async-event-processing-design.md b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
index d6510dc3cc5..eb3e193659a 100644
--- a/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
+++ b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
@@ -49,6 +49,16 @@ The async processing queue is active only when `enableAsyncProcessing=true`. Its
 
 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:
diff --git a/sentry/api/sentry.api b/sentry/api/sentry.api
index b807f70a88d..5944f33b767 100644
--- a/sentry/api/sentry.api
+++ b/sentry/api/sentry.api
@@ -475,6 +475,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 {
@@ -526,6 +531,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;
@@ -545,6 +551,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
@@ -3747,6 +3754,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
@@ -3810,6 +3818,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
diff --git a/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java b/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java
new file mode 100644
index 00000000000..f4f5eedcb21
--- /dev/null
+++ b/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;
+    }
+  }
+}
diff --git a/sentry/src/main/java/io/sentry/EventProcessor.java b/sentry/src/main/java/io/sentry/EventProcessor.java
index 928fd022095..3ab654ef829 100644
--- a/sentry/src/main/java/io/sentry/EventProcessor.java
+++ b/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.
    *
diff --git a/sentry/src/main/java/io/sentry/ExternalOptions.java b/sentry/src/main/java/io/sentry/ExternalOptions.java
index 4e44ea422ec..159548ac565 100644
--- a/sentry/src/main/java/io/sentry/ExternalOptions.java
+++ b/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;
   }
diff --git a/sentry/src/main/java/io/sentry/SentryClient.java b/sentry/src/main/java/io/sentry/SentryClient.java
index 92037f6690b..46bd720c2e7 100644
--- a/sentry/src/main/java/io/sentry/SentryClient.java
+++ b/sentry/src/main/java/io/sentry/SentryClient.java
@@ -44,6 +44,7 @@ public final class SentryClient implements ISentryClient {
   private final @NotNull SortBreadcrumbsByDate sortBreadcrumbsByDate = new SortBreadcrumbsByDate();
   private final @NotNull ILoggerBatchProcessor loggerBatchProcessor;
   private final @NotNull IMetricsBatchProcessor metricsBatchProcessor;
+  private final @NotNull AsyncEventProcessingExecutor asyncEventProcessingExecutor;
 
   @Override
   public boolean isEnabled() {
@@ -74,6 +75,8 @@ public SentryClient(final @NotNull SentryOptions options) {
     } else {
       metricsBatchProcessor = NoOpMetricsBatchProcessor.getInstance();
     }
+    asyncEventProcessingExecutor =
+        new AsyncEventProcessingExecutor(options.getMaxQueueSize(), options.getLogger());
   }
 
   private boolean shouldApplyScopeData(
@@ -168,6 +171,41 @@ private boolean shouldApplyScopeData(final @NotNull CheckIn event, final @NotNul
 
     event = processEvent(event, hint, options.getEventProcessors());
 
+    if (event == null) {
+      return SentryId.EMPTY_ID;
+    }
+
+    final SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
+    final @NotNull SentryEvent finalEvent = event;
+    final @NotNull Hint finalHint = hint;
+    if (options.isEnableAsyncProcessing()) {
+      final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(finalHint);
+      if (!asyncEventProcessingExecutor.submit(
+          () -> captureEventAfterProcessing(finalEvent, finalHint, scope, applyScopedProcessors))) {
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Error);
+        return SentryId.EMPTY_ID;
+      }
+      return sentryId;
+    }
+
+    return captureEventAfterProcessing(event, hint, scope, HintUtils.shouldApplyScopeData(hint));
+  }
+
+  private @NotNull SentryId captureEventAfterProcessing(
+      @NotNull SentryEvent event,
+      final @NotNull Hint hint,
+      final @Nullable IScope scope,
+      final boolean applyScopedProcessors) {
+    if (applyScopedProcessors && scope != null) {
+      event = processEventAsync(event, hint, scope.getEventProcessors(), DataCategory.Error);
+    }
+
+    if (event != null) {
+      event = processEventAsync(event, hint, options.getEventProcessors(), DataCategory.Error);
+    }
+
     if (event != null) {
       event = executeBeforeSend(event, hint);
 
@@ -183,10 +221,6 @@ private boolean shouldApplyScopeData(final @NotNull CheckIn event, final @NotNul
       event = EventSizeLimitingUtils.limitEventSize(event, hint, options);
     }
 
-    if (event == null) {
-      return SentryId.EMPTY_ID;
-    }
-
     @Nullable
     Session sessionBeforeUpdate =
         scope != null ? scope.withSession((@Nullable Session session) -> {}) : null;
@@ -339,6 +373,30 @@ private void finalizeTransaction(final @NotNull IScope scope, final @NotNull Hin
 
     event = processReplayEvent(event, hint, options.getEventProcessors());
 
+    if (event == null) {
+      return SentryId.EMPTY_ID;
+    }
+
+    final @NotNull SentryReplayEvent finalEvent = event;
+    final @NotNull Hint finalHint = hint;
+    if (options.isEnableAsyncProcessing()) {
+      if (!asyncEventProcessingExecutor.submit(
+          () -> captureReplayEventAfterProcessing(finalEvent, scope, finalHint))) {
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Replay);
+        return SentryId.EMPTY_ID;
+      }
+      return sentryId;
+    }
+
+    return captureReplayEventAfterProcessing(event, scope, hint);
+  }
+
+  private @NotNull SentryId captureReplayEventAfterProcessing(
+      @NotNull SentryReplayEvent event, final @Nullable IScope scope, final @NotNull Hint hint) {
+    event = processReplayEventAsync(event, hint, options.getEventProcessors());
+
     if (event != null) {
       event = executeBeforeSendReplay(event, hint);
 
@@ -354,6 +412,7 @@ private void finalizeTransaction(final @NotNull IScope scope, final @NotNull Hin
       return SentryId.EMPTY_ID;
     }
 
+    SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
     try {
       final @Nullable TraceContext traceContext = getTraceContext(scope, hint, event, null);
       final boolean cleanupReplayFolder = HintUtils.hasType(hint, Backfillable.class);
@@ -734,6 +793,224 @@ private SentryEvent processFeedbackEvent(
     return feedbackEvent;
   }
 
+  @Nullable
+  private SentryEvent processEventAsync(
+      @NotNull SentryEvent event,
+      final @NotNull Hint hint,
+      final @NotNull List<EventProcessor> eventProcessors,
+      final @NotNull DataCategory dataCategory) {
+    for (final EventProcessor processor : eventProcessors) {
+      try {
+        final boolean isBackfillingProcessor = processor instanceof BackfillingEventProcessor;
+        final boolean isBackfillable = HintUtils.hasType(hint, Backfillable.class);
+        if (isBackfillable && isBackfillingProcessor) {
+          event = processor.processAsync(event, hint);
+        } else if (!isBackfillable && !isBackfillingProcessor) {
+          event = processor.processAsync(event, hint);
+        }
+      } catch (Throwable e) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                e,
+                "An exception occurred while async processing event by processor: %s",
+                processor.getClass().getName());
+      }
+
+      if (event == null) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "Event was dropped by an async processor: %s",
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, dataCategory);
+        break;
+      }
+    }
+    return event;
+  }
+
+  @Nullable
+  private SentryLogEvent processLogEventAsync(
+      @NotNull SentryLogEvent event, final @NotNull List<EventProcessor> eventProcessors) {
+    for (final EventProcessor processor : eventProcessors) {
+      try {
+        event = processor.processAsync(event);
+      } catch (Throwable e) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                e,
+                "An exception occurred while async processing log event by processor: %s",
+                processor.getClass().getName());
+      }
+
+      if (event == null) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "Log event was dropped by an async processor: %s",
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, DataCategory.LogItem);
+        break;
+      }
+    }
+    return event;
+  }
+
+  @Nullable
+  private SentryMetricsEvent processMetricsEventAsync(
+      @NotNull SentryMetricsEvent event,
+      final @NotNull List<EventProcessor> eventProcessors,
+      final @NotNull Hint hint) {
+    for (final EventProcessor processor : eventProcessors) {
+      try {
+        event = processor.processAsync(event, hint);
+      } catch (Throwable e) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                e,
+                "An exception occurred while async processing metrics event by processor: %s",
+                processor.getClass().getName());
+      }
+
+      if (event == null) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "Metrics event was dropped by an async processor: %s",
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, DataCategory.TraceMetric);
+        break;
+      }
+    }
+    return event;
+  }
+
+  private @Nullable SentryTransaction processTransactionAsync(
+      @NotNull SentryTransaction transaction,
+      final @NotNull Hint hint,
+      final @NotNull List<EventProcessor> eventProcessors) {
+    for (final EventProcessor processor : eventProcessors) {
+      final int spanCountBeforeProcessor = transaction.getSpans().size();
+      try {
+        transaction = processor.processAsync(transaction, hint);
+      } catch (Throwable e) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                e,
+                "An exception occurred while async processing transaction by processor: %s",
+                processor.getClass().getName());
+      }
+      final int spanCountAfterProcessor = transaction == null ? 0 : transaction.getSpans().size();
+
+      if (transaction == null) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "Transaction was dropped by an async processor: %s",
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, DataCategory.Transaction);
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(
+                DiscardReason.EVENT_PROCESSOR, DataCategory.Span, spanCountBeforeProcessor + 1);
+        break;
+      } else if (spanCountAfterProcessor < spanCountBeforeProcessor) {
+        final int droppedSpanCount = spanCountBeforeProcessor - spanCountAfterProcessor;
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "%d spans were dropped by an async processor: %s",
+                droppedSpanCount,
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, DataCategory.Span, droppedSpanCount);
+      }
+    }
+    return transaction;
+  }
+
+  @Nullable
+  private SentryReplayEvent processReplayEventAsync(
+      @NotNull SentryReplayEvent replayEvent,
+      final @NotNull Hint hint,
+      final @NotNull List<EventProcessor> eventProcessors) {
+    for (final EventProcessor processor : eventProcessors) {
+      try {
+        replayEvent = processor.processAsync(replayEvent, hint);
+      } catch (Throwable e) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                e,
+                "An exception occurred while async processing replay event by processor: %s",
+                processor.getClass().getName());
+      }
+
+      if (replayEvent == null) {
+        options
+            .getLogger()
+            .log(
+                SentryLevel.DEBUG,
+                "Replay event was dropped by an async processor: %s",
+                processor.getClass().getName());
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.EVENT_PROCESSOR, DataCategory.Replay);
+        break;
+      }
+    }
+    return replayEvent;
+  }
+
+  @Nullable
+  private SentryEvent processFeedbackEventAsync(
+      @NotNull SentryEvent feedbackEvent,
+      final @NotNull Hint hint,
+      final @NotNull List<EventProcessor> eventProcessors) {
+    return processEventAsync(feedbackEvent, hint, eventProcessors, DataCategory.Feedback);
+  }
+
+  private void recordLostTransaction(
+      final @NotNull DiscardReason discardReason, final @NotNull SentryTransaction transaction) {
+    options.getClientReportRecorder().recordLostEvent(discardReason, DataCategory.Transaction);
+    options
+        .getClientReportRecorder()
+        .recordLostEvent(discardReason, DataCategory.Span, transaction.getSpans().size() + 1);
+  }
+
+  private void recordLostLog(
+      final @NotNull DiscardReason discardReason, final @NotNull SentryLogEvent logEvent) {
+    options.getClientReportRecorder().recordLostEvent(discardReason, DataCategory.LogItem);
+    final long logEventNumberOfBytes =
+        JsonSerializationUtils.byteSizeOf(options.getSerializer(), options.getLogger(), logEvent);
+    options
+        .getClientReportRecorder()
+        .recordLostEvent(discardReason, DataCategory.LogByte, logEventNumberOfBytes);
+  }
+
   @Deprecated
   @Override
   public void captureUserFeedback(final @NotNull UserFeedback userFeedback) {
@@ -1048,6 +1325,54 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
       return SentryId.EMPTY_ID;
     }
 
+    final @NotNull SentryTransaction finalTransaction = transaction;
+    final @NotNull Hint finalHint = hint;
+    final @Nullable TraceContext finalTraceContext = traceContext;
+    if (options.isEnableAsyncProcessing()) {
+      if (!asyncEventProcessingExecutor.submit(
+          () ->
+              captureTransactionAfterProcessing(
+                  finalTransaction,
+                  finalTraceContext,
+                  scope,
+                  finalHint,
+                  profilingTraceData,
+                  HintUtils.shouldApplyScopeData(finalHint)))) {
+        recordLostTransaction(DiscardReason.QUEUE_OVERFLOW, transaction);
+        return SentryId.EMPTY_ID;
+      }
+      return sentryId;
+    }
+
+    return captureTransactionAfterProcessing(
+        transaction,
+        traceContext,
+        scope,
+        hint,
+        profilingTraceData,
+        HintUtils.shouldApplyScopeData(hint));
+  }
+
+  private @NotNull SentryId captureTransactionAfterProcessing(
+      @NotNull SentryTransaction transaction,
+      final @Nullable TraceContext traceContext,
+      final @Nullable IScope scope,
+      final @NotNull Hint hint,
+      final @Nullable ProfilingTraceData profilingTraceData,
+      final boolean applyScopedProcessors) {
+    if (applyScopedProcessors && scope != null) {
+      transaction = processTransactionAsync(transaction, hint, scope.getEventProcessors());
+    }
+
+    if (transaction != null) {
+      transaction = processTransactionAsync(transaction, hint, options.getEventProcessors());
+    }
+
+    if (transaction == null) {
+      options.getLogger().log(SentryLevel.DEBUG, "Transaction was dropped by Event processors.");
+      return SentryId.EMPTY_ID;
+    }
+
     final int spanCountBeforeCallback = transaction.getSpans().size();
     transaction = executeBeforeSendTransaction(transaction, hint);
     final int spanCountAfterCallback = transaction == null ? 0 : transaction.getSpans().size();
@@ -1079,6 +1404,8 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
           .recordLostEvent(DiscardReason.BEFORE_SEND, DataCategory.Span, droppedSpanCount);
     }
 
+    SentryId sentryId =
+        transaction.getEventId() != null ? transaction.getEventId() : SentryId.EMPTY_ID;
     try {
       final SentryEnvelope envelope =
           buildEnvelope(
@@ -1250,6 +1577,45 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
 
     event = processFeedbackEvent(event, hint, options.getEventProcessors());
 
+    if (event == null) {
+      return SentryId.EMPTY_ID;
+    }
+
+    final SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
+    final @NotNull SentryEvent finalEvent = event;
+    final @NotNull Hint finalHint = hint;
+    if (options.isEnableAsyncProcessing()) {
+      final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(finalHint);
+      if (!asyncEventProcessingExecutor.submit(
+          () ->
+              captureFeedbackAfterProcessing(
+                  finalEvent, feedback, finalHint, scope, applyScopedProcessors))) {
+        options
+            .getClientReportRecorder()
+            .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Feedback);
+        return SentryId.EMPTY_ID;
+      }
+      return sentryId;
+    }
+
+    return captureFeedbackAfterProcessing(
+        event, feedback, hint, scope, HintUtils.shouldApplyScopeData(hint));
+  }
+
+  private @NotNull SentryId captureFeedbackAfterProcessing(
+      @NotNull SentryEvent event,
+      final @NotNull Feedback feedback,
+      final @NotNull Hint hint,
+      final @NotNull IScope scope,
+      final boolean applyScopedProcessors) {
+    if (applyScopedProcessors) {
+      event = processFeedbackEventAsync(event, hint, scope.getEventProcessors());
+    }
+
+    if (event != null) {
+      event = processFeedbackEventAsync(event, hint, options.getEventProcessors());
+    }
+
     if (event != null) {
       event = executeBeforeSendFeedback(event, hint);
 
@@ -1265,10 +1631,7 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
       return SentryId.EMPTY_ID;
     }
 
-    SentryId sentryId = SentryId.EMPTY_ID;
-    if (event.getEventId() != null) {
-      sentryId = event.getEventId();
-    }
+    SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
 
     // If feedback already has a replayId, we don't want to overwrite it.
     if (feedback.getReplayId() == null) {
@@ -1352,22 +1715,37 @@ public void captureLog(@Nullable SentryLogEvent logEvent, @Nullable IScope scope
       }
     }
 
+    if (logEvent != null) {
+      final @NotNull SentryLogEvent finalLogEvent = logEvent;
+      if (options.isEnableAsyncProcessing()) {
+        if (!asyncEventProcessingExecutor.submit(
+            () -> captureLogAfterProcessing(finalLogEvent, scope))) {
+          recordLostLog(DiscardReason.QUEUE_OVERFLOW, finalLogEvent);
+        }
+        return;
+      }
+
+      captureLogAfterProcessing(logEvent, scope);
+    }
+  }
+
+  private void captureLogAfterProcessing(
+      @NotNull SentryLogEvent logEvent, final @Nullable IScope scope) {
+    if (scope != null) {
+      logEvent = processLogEventAsync(logEvent, scope.getEventProcessors());
+    }
+
+    if (logEvent != null) {
+      logEvent = processLogEventAsync(logEvent, options.getEventProcessors());
+    }
+
     if (logEvent != null) {
       final @NotNull SentryLogEvent tmpLogEvent = logEvent;
       logEvent = executeBeforeSendLog(logEvent);
 
       if (logEvent == null) {
         options.getLogger().log(SentryLevel.DEBUG, "Log Event was dropped by beforeSendLog");
-        options
-            .getClientReportRecorder()
-            .recordLostEvent(DiscardReason.BEFORE_SEND, DataCategory.LogItem);
-        final @NotNull long logEventNumberOfBytes =
-            JsonSerializationUtils.byteSizeOf(
-                options.getSerializer(), options.getLogger(), tmpLogEvent);
-        options
-            .getClientReportRecorder()
-            .recordLostEvent(
-                DiscardReason.BEFORE_SEND, DataCategory.LogByte, logEventNumberOfBytes);
+        recordLostLog(DiscardReason.BEFORE_SEND, tmpLogEvent);
         return;
       }
 
@@ -1415,6 +1793,35 @@ public void captureMetric(
       }
     }
 
+    if (metricsEvent != null) {
+      final @NotNull SentryMetricsEvent finalMetricsEvent = metricsEvent;
+      final @NotNull Hint finalHint = hint;
+      if (options.isEnableAsyncProcessing()) {
+        if (!asyncEventProcessingExecutor.submit(
+            () -> captureMetricAfterProcessing(finalMetricsEvent, scope, finalHint))) {
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.TraceMetric);
+        }
+        return;
+      }
+
+      captureMetricAfterProcessing(metricsEvent, scope, hint);
+    }
+  }
+
+  private void captureMetricAfterProcessing(
+      @NotNull SentryMetricsEvent metricsEvent,
+      final @Nullable IScope scope,
+      final @NotNull Hint hint) {
+    if (scope != null) {
+      metricsEvent = processMetricsEventAsync(metricsEvent, scope.getEventProcessors(), hint);
+    }
+
+    if (metricsEvent != null) {
+      metricsEvent = processMetricsEventAsync(metricsEvent, options.getEventProcessors(), hint);
+    }
+
     if (metricsEvent != null) {
       final @NotNull SentryMetricsEvent tmpMetricsEvent = metricsEvent;
       metricsEvent = executeBeforeSendMetric(metricsEvent, hint);
@@ -1788,7 +2195,10 @@ public void close() {
   public void close(final boolean isRestarting) {
     options.getLogger().log(SentryLevel.INFO, "Closing SentryClient.");
     try {
-      flush(isRestarting ? 0 : options.getShutdownTimeoutMillis());
+      final long timeoutMillis = isRestarting ? 0 : options.getShutdownTimeoutMillis();
+      final long deadline = System.currentTimeMillis() + timeoutMillis;
+      flush(timeoutMillis);
+      asyncEventProcessingExecutor.close(remainingTimeout(deadline));
       loggerBatchProcessor.close(isRestarting);
       metricsBatchProcessor.close(isRestarting);
       transport.close(isRestarting);
@@ -1817,9 +2227,15 @@ public void close(final boolean isRestarting) {
 
   @Override
   public void flush(final long timeoutMillis) {
-    loggerBatchProcessor.flush(timeoutMillis);
-    metricsBatchProcessor.flush(timeoutMillis);
-    transport.flush(timeoutMillis);
+    final long deadline = System.currentTimeMillis() + timeoutMillis;
+    asyncEventProcessingExecutor.waitTillIdle(timeoutMillis);
+    loggerBatchProcessor.flush(remainingTimeout(deadline));
+    metricsBatchProcessor.flush(remainingTimeout(deadline));
+    transport.flush(remainingTimeout(deadline));
+  }
+
+  private long remainingTimeout(final long deadline) {
+    return Math.max(0, deadline - System.currentTimeMillis());
   }
 
   @Override
diff --git a/sentry/src/main/java/io/sentry/SentryOptions.java b/sentry/src/main/java/io/sentry/SentryOptions.java
index dc9d9521dbb..26907158d3a 100644
--- a/sentry/src/main/java/io/sentry/SentryOptions.java
+++ b/sentry/src/main/java/io/sentry/SentryOptions.java
@@ -374,6 +374,9 @@ public class SentryOptions {
    */
   private boolean enableDeduplication = true;
 
+  /** Whether EventProcessor.processAsync and beforeSend callbacks run off the caller thread. */
+  private boolean enableAsyncProcessing = false;
+
   /**
    * Enables event size limiting with {@link EventSizeLimitingEventProcessor}. When enabled, events
    * exceeding 1MB will have breadcrumbs and stack frames reduced to stay under the limit.
@@ -1883,6 +1886,24 @@ public void setEnableDeduplication(final boolean enableDeduplication) {
     this.enableDeduplication = enableDeduplication;
   }
 
+  /**
+   * Returns whether async event processing is enabled.
+   *
+   * @return true if async event processing is enabled, false otherwise
+   */
+  public boolean isEnableAsyncProcessing() {
+    return enableAsyncProcessing;
+  }
+
+  /**
+   * Enables or disables async event processing.
+   *
+   * @param enableAsyncProcessing true if enabled, false otherwise
+   */
+  public void setEnableAsyncProcessing(final boolean enableAsyncProcessing) {
+    this.enableAsyncProcessing = enableAsyncProcessing;
+  }
+
   /**
    * Returns if event size limiting is enabled.
    *
@@ -3608,6 +3629,9 @@ public void merge(final @NotNull ExternalOptions options) {
     if (options.getEnableDeduplication() != null) {
       setEnableDeduplication(options.getEnableDeduplication());
     }
+    if (options.isEnableAsyncProcessing() != null) {
+      setEnableAsyncProcessing(options.isEnableAsyncProcessing());
+    }
     if (options.getSendClientReports() != null) {
       setSendClientReports(options.getSendClientReports());
     }
diff --git a/sentry/src/test/java/io/sentry/ExternalOptionsTest.kt b/sentry/src/test/java/io/sentry/ExternalOptionsTest.kt
index fee707d31f3..eeb1169a279 100644
--- a/sentry/src/test/java/io/sentry/ExternalOptionsTest.kt
+++ b/sentry/src/test/java/io/sentry/ExternalOptionsTest.kt
@@ -123,6 +123,18 @@ class ExternalOptionsTest {
     }
   }
 
+  @Test
+  fun `creates options with enableAsyncProcessing using external properties`() {
+    withPropertiesFile("enable-async-processing=true") {
+      assertNotNull(it.isEnableAsyncProcessing) { assertTrue(it) }
+    }
+  }
+
+  @Test
+  fun `creates options with enableAsyncProcessing set to null when not set`() {
+    withPropertiesFile { assertNull(it.isEnableAsyncProcessing) }
+  }
+
   @Test
   fun `creates options with sendClientReports using external properties`() {
     withPropertiesFile("send-client-reports=false") {
diff --git a/sentry/src/test/java/io/sentry/SentryClientTest.kt b/sentry/src/test/java/io/sentry/SentryClientTest.kt
index 02623556498..7e80e0cadc7 100644
--- a/sentry/src/test/java/io/sentry/SentryClientTest.kt
+++ b/sentry/src/test/java/io/sentry/SentryClientTest.kt
@@ -46,6 +46,8 @@ import java.nio.file.Files
 import java.util.Arrays
 import java.util.LinkedList
 import java.util.UUID
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFails
@@ -185,9 +187,15 @@ class SentryClientTest {
     assertTrue(sut.isEnabled)
     sut.close(false)
     assertNotEquals(0, fixture.sentryOptions.shutdownTimeoutMillis)
-    verify(fixture.transport).flush(eq(fixture.sentryOptions.shutdownTimeoutMillis))
-    verify(fixture.loggerBatchProcessor).flush(eq(fixture.sentryOptions.shutdownTimeoutMillis))
-    verify(fixture.metricsBatchProcessor).flush(eq(fixture.sentryOptions.shutdownTimeoutMillis))
+    val transportFlushTimeout = argumentCaptor<Long>()
+    verify(fixture.transport).flush(transportFlushTimeout.capture())
+    assertTrue(transportFlushTimeout.firstValue in 0..fixture.sentryOptions.shutdownTimeoutMillis)
+    val loggerFlushTimeout = argumentCaptor<Long>()
+    verify(fixture.loggerBatchProcessor).flush(loggerFlushTimeout.capture())
+    assertTrue(loggerFlushTimeout.firstValue in 0..fixture.sentryOptions.shutdownTimeoutMillis)
+    val metricsFlushTimeout = argumentCaptor<Long>()
+    verify(fixture.metricsBatchProcessor).flush(metricsFlushTimeout.capture())
+    assertTrue(metricsFlushTimeout.firstValue in 0..fixture.sentryOptions.shutdownTimeoutMillis)
     verify(fixture.transport).close(eq(false))
     verify(fixture.loggerBatchProcessor).close(eq(false))
     verify(fixture.metricsBatchProcessor).close(eq(false))
@@ -236,6 +244,110 @@ class SentryClientTest {
     assertTrue(invoked)
   }
 
+  @Test
+  fun `EventProcessor processAsync defaults return input`() {
+    val processor = object : EventProcessor {}
+    val hint = Hint()
+    val event = SentryEvent()
+    val transaction = SentryTransaction(fixture.sentryTracer)
+    val replay = SentryReplayEvent()
+    val log = SentryLogEvent(SentryId(), 1.0, "body", SentryLogLevel.INFO)
+    val metrics = SentryMetricsEvent(SentryId(), 1.0, "metric", "counter", 1.0)
+
+    assertSame(event, processor.processAsync(event, hint))
+    assertSame(transaction, processor.processAsync(transaction, hint))
+    assertSame(replay, processor.processAsync(replay, hint))
+    assertSame(log, processor.processAsync(log))
+    assertSame(metrics, processor.processAsync(metrics, hint))
+  }
+
+  @Test
+  fun `when async processing is disabled, processAsync runs inline before beforeSend`() {
+    val order = mutableListOf<String>()
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun process(event: SentryEvent, hint: Hint): SentryEvent? {
+          order.add("process")
+          return event
+        }
+
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          order.add("processAsync")
+          return event
+        }
+      }
+    )
+    fixture.sentryOptions.setBeforeSend { event, _ ->
+      order.add("beforeSend")
+      event
+    }
+
+    val sut = fixture.getSut()
+    sut.captureEvent(SentryEvent())
+
+    assertEquals(listOf("process", "processAsync", "beforeSend"), order)
+  }
+
+  @Test
+  fun `when async processing is enabled, captureEvent returns after task is accepted`() {
+    val asyncStarted = CountDownLatch(1)
+    val continueAsync = CountDownLatch(1)
+    val eventId = SentryId()
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          asyncStarted.countDown()
+          assertTrue(continueAsync.await(5, TimeUnit.SECONDS))
+          return event
+        }
+      }
+    )
+    val sut = fixture.getSut()
+    val event = SentryEvent().apply { this.eventId = eventId }
+
+    val actual = sut.captureEvent(event)
+
+    assertEquals(eventId, actual)
+    assertTrue(asyncStarted.await(5, TimeUnit.SECONDS))
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+
+    continueAsync.countDown()
+    sut.flush(5000)
+
+    verify(fixture.transport).send(any(), anyOrNull())
+  }
+
+  @Test
+  fun `when async processing queue is full, captureEvent returns empty id and records lost event`() {
+    val asyncStarted = CountDownLatch(1)
+    val continueAsync = CountDownLatch(1)
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    fixture.sentryOptions.maxQueueSize = 1
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          asyncStarted.countDown()
+          assertTrue(continueAsync.await(5, TimeUnit.SECONDS))
+          return event
+        }
+      }
+    )
+    val sut = fixture.getSut()
+
+    assertNotEquals(SentryId.EMPTY_ID, sut.captureEvent(SentryEvent()))
+    assertTrue(asyncStarted.await(5, TimeUnit.SECONDS))
+    assertEquals(SentryId.EMPTY_ID, sut.captureEvent(SentryEvent()))
+
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.QUEUE_OVERFLOW.reason, DataCategory.Error.category, 1)),
+    )
+
+    continueAsync.countDown()
+    sut.flush(5000)
+  }
+
   @Test
   fun `when beforeSend is returns null, event is dropped`() {
     fixture.sentryOptions.setBeforeSend { _: SentryEvent, _: Any? -> null }
@@ -2664,6 +2776,12 @@ class SentryClientTest {
     whenever(globalEventProcessorMock.process(any<SentryEvent>(), anyOrNull())).doAnswer {
       it.arguments.first() as SentryEvent
     }
+    whenever(scopedEventProcessorMock.processAsync(any<SentryEvent>(), anyOrNull())).doAnswer {
+      it.arguments.first() as SentryEvent
+    }
+    whenever(globalEventProcessorMock.processAsync(any<SentryEvent>(), anyOrNull())).doAnswer {
+      it.arguments.first() as SentryEvent
+    }
     whenever(beforeSendMock.execute(any(), anyOrNull())).doAnswer {
       it.arguments.first() as SentryEvent
     }
@@ -2708,6 +2826,12 @@ class SentryClientTest {
     whenever(globalEventProcessorMock.process(any<SentryEvent>(), anyOrNull())).doAnswer {
       it.arguments.first() as SentryEvent
     }
+    whenever(scopedEventProcessorMock.processAsync(any<SentryEvent>(), anyOrNull())).doAnswer {
+      it.arguments.first() as SentryEvent
+    }
+    whenever(globalEventProcessorMock.processAsync(any<SentryEvent>(), anyOrNull())).doAnswer {
+      it.arguments.first() as SentryEvent
+    }
     whenever(beforeSendMock.execute(any(), anyOrNull())).thenReturn(null)
 
     val sut = fixture.getSut { options ->
diff --git a/sentry/src/test/java/io/sentry/SentryOptionsTest.kt b/sentry/src/test/java/io/sentry/SentryOptionsTest.kt
index f1ee97459a1..fe44cea9de8 100644
--- a/sentry/src/test/java/io/sentry/SentryOptionsTest.kt
+++ b/sentry/src/test/java/io/sentry/SentryOptionsTest.kt
@@ -49,6 +49,18 @@ class SentryOptionsTest {
     assertFalse(SentryOptions().isDebug)
   }
 
+  @Test
+  fun `when options is initialized, async processing is false`() {
+    assertFalse(SentryOptions().isEnableAsyncProcessing)
+  }
+
+  @Test
+  fun `when enableAsyncProcessing is set, overrides default`() {
+    val options = SentryOptions()
+    options.isEnableAsyncProcessing = true
+    assertTrue(options.isEnableAsyncProcessing)
+  }
+
   @Test
   fun `when options is initialized, integrations contain UncaughtExceptionHandlerIntegration`() {
     assertTrue(SentryOptions().integrations.any { it is UncaughtExceptionHandlerIntegration })
@@ -399,6 +411,7 @@ class SentryOptionsTest {
     externalOptions.ignoredTransactions = listOf("transactionName1", "transaction-name-B")
     externalOptions.ignoredErrors = listOf("Some error", "Another .*")
     externalOptions.isEnableBackpressureHandling = false
+    externalOptions.isEnableAsyncProcessing = true
     externalOptions.isEnableDatabaseTransactionTracing = true
     externalOptions.isEnableCacheTracing = true
     externalOptions.maxRequestBodySize = SentryOptions.RequestSize.MEDIUM
@@ -465,6 +478,7 @@ class SentryOptionsTest {
       options.ignoredErrors,
     )
     assertFalse(options.isEnableBackpressureHandling)
+    assertTrue(options.isEnableAsyncProcessing)
     assertTrue(options.isEnableDatabaseTransactionTracing)
     assertTrue(options.isEnableCacheTracing)
     assertTrue(options.isForceInit)

From 513b42ec7cc4e9f48a864fd2a744f25675e051ab Mon Sep 17 00:00:00 2001
From: Alexander Dinauer <adinauer@users.noreply.github.com>
Date: Wed, 17 Jun 2026 14:19:37 +0200
Subject: [PATCH 3/4] changelog

---
 CHANGELOG.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d826db437c5..79803f76a92 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -286,6 +286,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

From 3a02b30d0092d95716c53e895b01300a3f95d2b9 Mon Sep 17 00:00:00 2001
From: Nelson Osacky <nelson.osacky@sentry.io>
Date: Thu, 30 Jul 2026 16:46:08 +0200
Subject: [PATCH 4/4] fix(core): Make async event processing safe for crash and
 shutdown paths

The async event processing stage moved work to another thread without
establishing what state that work may touch. Four issues followed.

close() flushed downstream queues before quiescing the async stage, so
tasks finishing afterwards handed envelopes to a transport that was
already closing, and anything still queued when the timeout expired was
discarded with no client report. Quiesce the async stage first and record
queue_overflow for work that could not be drained.

Captures whose hint coordinates across threads are no longer deferred.
DiskFlushNotification captures (uncaught exceptions, ANRv2, tombstones)
block the caller until the transport reports the event flushed to disk;
queueing put the fatal event behind everything already enqueued, so the
caller's flush timeout could expire before it was ever sent. Backfillable
and Cached captures carry retry/delete state the caller reads as soon as
capture returns, and TransactionEnd captures force-finish the running
transaction before the process dies.

The scope passed to capture* is a live view over the caller's
thread-local scope, so the async stage could read tags, sessions, and
transactions belonging to unrelated later work. It is now cloned at
submit time.

Hint left its attachment fields unsynchronized while both threads could
still reach them. The attachment list is now guarded by the existing lock
and the single-reference fields are volatile.

Also stop holding the executor monitor across awaitTermination, which
could stall a capture on an app thread for the whole shutdown timeout.
---
 CHANGELOG.md                                  |   8 +-
 ...026-06-15-async-event-processing-design.md |  18 ++-
 .../sentry/AsyncEventProcessingExecutor.java  | 113 ++++++++++---
 sentry/src/main/java/io/sentry/Hint.java      |  35 ++--
 .../src/main/java/io/sentry/SentryClient.java | 123 ++++++++++----
 .../AsyncEventProcessingExecutorTest.kt       | 151 ++++++++++++++++++
 .../test/java/io/sentry/SentryClientTest.kt   | 133 +++++++++++++++
 7 files changed, 518 insertions(+), 63 deletions(-)
 create mode 100644 sentry/src/test/java/io/sentry/AsyncEventProcessingExecutorTest.kt

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 79803f76a92..da80cdbebb0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+### Features
+
+- Add opt-in async event processing ([#5558](https://github.com/getsentry/sentry-java/pull/5558))
+  - Disabled by default; opt in via `options.isEnableAsyncProcessing = true` or the external option `enable-async-processing`
+  - Moves `EventProcessor.processAsync(...)` and the matching `beforeSend*` callback off the caller thread onto a dedicated bounded queue
+  - Crash, ANR, cached, and backfillable captures always stay synchronous so their flush-to-disk and retry handshakes keep working
+
 ### Fixes
 
 - Prevent inflated cold app start when the OS spawns the process in the background (e.g. FCM push) on API 35+ ([#5841](https://github.com/getsentry/sentry-java/pull/5841))
@@ -286,7 +293,6 @@
 
 ### 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
diff --git a/docs/superpowers/specs/2026-06-15-async-event-processing-design.md b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
index eb3e193659a..2965eb1c307 100644
--- a/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
+++ b/docs/superpowers/specs/2026-06-15-async-event-processing-design.md
@@ -82,6 +82,18 @@ Supported paths:
 
 Sessions, check-ins, and profile chunks are unchanged.
 
+## Captures That Always Stay Synchronous
+
+Some captures coordinate across threads and must not be deferred, even when `enableAsyncProcessing=true`. These run the late processing stage inline on the caller thread:
+
+- Hints implementing `DiskFlushNotification` (uncaught exceptions, ANRv2, tombstones). The caller blocks until the transport reports the event flushed to disk before letting the process die. Queueing would put the fatal event behind everything already enqueued, so the caller's flush timeout can expire before it is ever sent.
+- Hints implementing `Backfillable` or `Cached` (events replayed from the outbox or cache). The hint carries retry/delete state that the caller reads the moment capture returns, and the envelope file is deleted based on it.
+- Hints implementing `TransactionEnd` (ANR). The capture force-finishes the running transaction so it is persisted before the process dies; deferring that races the process going away.
+
+## Scope Snapshotting
+
+The scope handed to `capture*` is a live `CombinedScopeView` over the caller's thread-local scope. When work is deferred, the scope is cloned at submit time so the async stage sees the state as of the capture call. Without this, a caller that pushes or pops scopes, changes tags, or starts a new transaction before the async stage runs would have that unrelated state attributed to the event.
+
 ## 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.
@@ -110,7 +122,7 @@ Transaction accounting must remain consistent:
 
 `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.
+`SentryClient.close()` quiesces the async processing queue *before* draining downstream, so no task can hand an envelope to a transport that is already closing. Work still queued when the shutdown timeout expires is discarded and recorded as `queue_overflow` for its data category rather than being lost silently. Closing remains bounded by existing shutdown and flush timeout behavior, with the async drain and the downstream flush sharing one deadline.
 
 ## Testing Plan
 
@@ -124,3 +136,7 @@ Transaction accounting must remain consistent:
 - 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.
+- Verify crash, cached, and backfillable hints bypass the queue and are sent on the caller thread.
+- Verify `close` stops accepting new async work before draining downstream, and records `queue_overflow` for work it could not drain.
+- Verify the async stage sees the scope as of the capture call, not later caller mutations.
+- Verify submitting does not block behind a draining `close`.
diff --git a/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java b/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java
index f4f5eedcb21..39bc0d1be3c 100644
--- a/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java
+++ b/sentry/src/main/java/io/sentry/AsyncEventProcessingExecutor.java
@@ -1,10 +1,13 @@
 package io.sentry;
 
 import io.sentry.transport.ReusableCountLatch;
+import io.sentry.util.AutoClosableReentrantLock;
+import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
 
@@ -14,7 +17,14 @@ final class AsyncEventProcessingExecutor {
   private final @NotNull ILogger logger;
   private final @NotNull ThreadPoolExecutor executor;
   private final @NotNull ReusableCountLatch unfinishedTasksCount = new ReusableCountLatch();
-  private boolean closed = false;
+  private final @NotNull AtomicBoolean closed = new AtomicBoolean(false);
+
+  /**
+   * Guards the admission check and the matching increment so concurrent submits cannot both observe
+   * the same free slot. Never held across a blocking call: a capture on an app thread must not
+   * stall behind a shutdown that is draining the queue.
+   */
+  private final @NotNull AutoClosableReentrantLock submitLock = new AutoClosableReentrantLock();
 
   AsyncEventProcessingExecutor(final int maxQueueSize, final @NotNull ILogger logger) {
     this.maxQueueSize = maxQueueSize;
@@ -29,21 +39,25 @@ final class AsyncEventProcessingExecutor {
             new AsyncEventProcessingThreadFactory());
   }
 
-  synchronized boolean submit(final @NotNull Runnable task) {
-    if (closed || unfinishedTasksCount.getCount() >= maxQueueSize) {
-      return false;
+  /**
+   * Accepts a task for async processing.
+   *
+   * @param task the processing to run off the caller thread
+   * @param onDropped invoked if the task is discarded during shutdown, so the caller can record the
+   *     matching client report. Never invoked when this method returns {@code false}; the caller
+   *     reports the rejection itself in that case.
+   * @return true if the task was accepted
+   */
+  boolean submit(final @NotNull Runnable task, final @NotNull Runnable onDropped) {
+    try (final @NotNull ISentryLifecycleToken ignored = submitLock.acquire()) {
+      if (closed.get() || unfinishedTasksCount.getCount() >= maxQueueSize) {
+        return false;
+      }
+      unfinishedTasksCount.increment();
     }
 
-    unfinishedTasksCount.increment();
     try {
-      executor.execute(
-          () -> {
-            try {
-              task.run();
-            } finally {
-              unfinishedTasksCount.decrement();
-            }
-          });
+      executor.execute(new AsyncTask(task, onDropped));
       return true;
     } catch (Throwable e) {
       unfinishedTasksCount.decrement();
@@ -52,28 +66,89 @@ synchronized boolean submit(final @NotNull Runnable task) {
     }
   }
 
+  boolean isClosed() {
+    return closed.get();
+  }
+
   void waitTillIdle(final long timeoutMillis) {
     try {
-      unfinishedTasksCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
+      if (!unfinishedTasksCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS)) {
+        logger.log(
+            SentryLevel.WARNING,
+            "Timed out waiting for async event processing queue to drain, %d task(s) still pending.",
+            unfinishedTasksCount.getCount());
+      }
     } catch (InterruptedException e) {
-      logger.log(SentryLevel.ERROR, "Failed to wait for async event processing queue to drain.", e);
+      logger.log(
+          SentryLevel.DEBUG,
+          "Interrupted while waiting for async event processing queue to drain.");
       Thread.currentThread().interrupt();
     }
   }
 
-  synchronized void close(final long timeoutMillis) {
-    closed = true;
+  /**
+   * Stops accepting new work and drains what was already accepted. Must run before downstream
+   * queues are flushed, otherwise tasks finishing here would hand envelopes to an already-drained
+   * transport.
+   */
+  void close(final long timeoutMillis) {
+    closed.set(true);
     executor.shutdown();
     try {
       if (!executor.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS)) {
-        executor.shutdownNow();
+        reportDropped(executor.shutdownNow());
       }
     } catch (InterruptedException e) {
-      executor.shutdownNow();
+      reportDropped(executor.shutdownNow());
       Thread.currentThread().interrupt();
     }
   }
 
+  /** Records a client report for every task that never got to run, so drops stay accounted for. */
+  private void reportDropped(final @NotNull List<Runnable> neverRun) {
+    if (neverRun.isEmpty()) {
+      return;
+    }
+    logger.log(
+        SentryLevel.WARNING,
+        "Dropping %d async event processing task(s) that did not finish before shutdown.",
+        neverRun.size());
+    for (final Runnable runnable : neverRun) {
+      if (runnable instanceof AsyncTask) {
+        ((AsyncTask) runnable).drop();
+      }
+    }
+  }
+
+  private final class AsyncTask implements Runnable {
+    private final @NotNull Runnable task;
+    private final @NotNull Runnable onDropped;
+
+    AsyncTask(final @NotNull Runnable task, final @NotNull Runnable onDropped) {
+      this.task = task;
+      this.onDropped = onDropped;
+    }
+
+    @Override
+    public void run() {
+      try {
+        task.run();
+      } finally {
+        unfinishedTasksCount.decrement();
+      }
+    }
+
+    void drop() {
+      try {
+        onDropped.run();
+      } catch (Throwable e) {
+        logger.log(SentryLevel.ERROR, "Failed to record dropped async event processing task.", e);
+      } finally {
+        unfinishedTasksCount.decrement();
+      }
+    }
+  }
+
   private static final class AsyncEventProcessingThreadFactory implements ThreadFactory {
     private int cnt;
 
diff --git a/sentry/src/main/java/io/sentry/Hint.java b/sentry/src/main/java/io/sentry/Hint.java
index 1e09dca5541..ee80346ab89 100644
--- a/sentry/src/main/java/io/sentry/Hint.java
+++ b/sentry/src/main/java/io/sentry/Hint.java
@@ -29,11 +29,14 @@ public final class Hint {
   private final @NotNull Map<String, Object> internalStorage = new HashMap<String, Object>();
   private final @NotNull List<Attachment> attachments = new ArrayList<>();
   private final @NotNull AutoClosableReentrantLock lock = new AutoClosableReentrantLock();
-  private @Nullable Attachment screenshot = null;
-  private @Nullable Attachment viewHierarchy = null;
-  private @Nullable Attachment threadDump = null;
-  private @Nullable Attachment tombstone = null;
-  private @Nullable ReplayRecording replayRecording = null;
+
+  // A Hint can be handed to another thread while the caller still holds a reference, e.g. when
+  // async event processing is enabled. These are volatile so that thread sees the latest write.
+  private volatile @Nullable Attachment screenshot = null;
+  private volatile @Nullable Attachment viewHierarchy = null;
+  private volatile @Nullable Attachment threadDump = null;
+  private volatile @Nullable Attachment tombstone = null;
+  private volatile @Nullable ReplayRecording replayRecording = null;
 
   public static @NotNull Hint withAttachment(@Nullable Attachment attachment) {
     @NotNull final Hint hint = new Hint();
@@ -82,27 +85,37 @@ public void remove(@NotNull String name) {
 
   public void addAttachment(@Nullable Attachment attachment) {
     if (attachment != null) {
-      attachments.add(attachment);
+      try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
+        attachments.add(attachment);
+      }
     }
   }
 
   public void addAttachments(@Nullable List<Attachment> attachments) {
     if (attachments != null) {
-      this.attachments.addAll(attachments);
+      try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
+        this.attachments.addAll(attachments);
+      }
     }
   }
 
   public @NotNull List<Attachment> getAttachments() {
-    return new ArrayList<>(attachments);
+    try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
+      return new ArrayList<>(attachments);
+    }
   }
 
   public void replaceAttachments(@Nullable List<Attachment> attachments) {
-    clearAttachments();
-    addAttachments(attachments);
+    try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
+      clearAttachments();
+      addAttachments(attachments);
+    }
   }
 
   public void clearAttachments() {
-    attachments.clear();
+    try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
+      attachments.clear();
+    }
   }
 
   /**
diff --git a/sentry/src/main/java/io/sentry/SentryClient.java b/sentry/src/main/java/io/sentry/SentryClient.java
index 46bd720c2e7..673aece04a4 100644
--- a/sentry/src/main/java/io/sentry/SentryClient.java
+++ b/sentry/src/main/java/io/sentry/SentryClient.java
@@ -91,6 +91,45 @@ private boolean shouldApplyScopeData(
     }
   }
 
+  /**
+   * Whether the late processing stage may run off the caller thread for this capture.
+   *
+   * <p>Async processing is skipped for hints that coordinate across threads even when the option is
+   * on:
+   *
+   * <ul>
+   *   <li>{@link DiskFlushNotification} — the caller blocks until the transport marks the event
+   *       flushed (crash, ANR, tombstone). Queueing puts the fatal event behind everything already
+   *       enqueued, so the caller's flush timeout can expire before it is ever sent.
+   *   <li>{@link Backfillable} and {@link Cached} — events replayed from disk. The hint carries
+   *       retry/delete state that the caller inspects the moment capture returns, and the envelope
+   *       file is deleted based on it.
+   *   <li>{@link TransactionEnd} — the capture force-finishes the running transaction so it is
+   *       persisted before the process dies (ANR). Deferring that races the process going away.
+   * </ul>
+   */
+  private boolean shouldProcessAsync(final @NotNull Hint hint) {
+    if (!options.isEnableAsyncProcessing()) {
+      return false;
+    }
+    return !HintUtils.hasType(hint, DiskFlushNotification.class)
+        && !HintUtils.hasType(hint, Backfillable.class)
+        && !HintUtils.hasType(hint, Cached.class)
+        && !HintUtils.hasType(hint, TransactionEnd.class);
+  }
+
+  /**
+   * Snapshots the scope so async processing sees the state as of the capture call.
+   *
+   * <p>The scope handed to {@code capture*} is a live {@link CombinedScopeView} over the caller's
+   * thread-local scope. By the time the async stage runs, the caller may have pushed or popped
+   * scopes, changed tags, or started a different transaction; reading it there would attribute
+   * unrelated state to this event.
+   */
+  private @Nullable IScope snapshotScope(final @Nullable IScope scope) {
+    return scope == null ? null : scope.clone();
+  }
+
   private boolean shouldApplyScopeData(final @NotNull CheckIn event, final @NotNull Hint hint) {
     if (HintUtils.shouldApplyScopeData(hint)) {
       return true;
@@ -177,11 +216,17 @@ private boolean shouldApplyScopeData(final @NotNull CheckIn event, final @NotNul
 
     final SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
     final @NotNull SentryEvent finalEvent = event;
-    final @NotNull Hint finalHint = hint;
-    if (options.isEnableAsyncProcessing()) {
-      final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(finalHint);
+    final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(hint);
+    if (shouldProcessAsync(hint)) {
+      final @NotNull Hint asyncHint = hint;
+      final @Nullable IScope asyncScope = snapshotScope(scope);
       if (!asyncEventProcessingExecutor.submit(
-          () -> captureEventAfterProcessing(finalEvent, finalHint, scope, applyScopedProcessors))) {
+          () ->
+              captureEventAfterProcessing(finalEvent, asyncHint, asyncScope, applyScopedProcessors),
+          () ->
+              options
+                  .getClientReportRecorder()
+                  .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Error))) {
         options
             .getClientReportRecorder()
             .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Error);
@@ -190,7 +235,7 @@ private boolean shouldApplyScopeData(final @NotNull CheckIn event, final @NotNul
       return sentryId;
     }
 
-    return captureEventAfterProcessing(event, hint, scope, HintUtils.shouldApplyScopeData(hint));
+    return captureEventAfterProcessing(event, hint, scope, applyScopedProcessors);
   }
 
   private @NotNull SentryId captureEventAfterProcessing(
@@ -378,10 +423,15 @@ private void finalizeTransaction(final @NotNull IScope scope, final @NotNull Hin
     }
 
     final @NotNull SentryReplayEvent finalEvent = event;
-    final @NotNull Hint finalHint = hint;
-    if (options.isEnableAsyncProcessing()) {
+    if (shouldProcessAsync(hint)) {
+      final @NotNull Hint asyncHint = hint;
+      final @Nullable IScope asyncScope = snapshotScope(scope);
       if (!asyncEventProcessingExecutor.submit(
-          () -> captureReplayEventAfterProcessing(finalEvent, scope, finalHint))) {
+          () -> captureReplayEventAfterProcessing(finalEvent, asyncScope, asyncHint),
+          () ->
+              options
+                  .getClientReportRecorder()
+                  .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Replay))) {
         options
             .getClientReportRecorder()
             .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Replay);
@@ -1326,18 +1376,21 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
     }
 
     final @NotNull SentryTransaction finalTransaction = transaction;
-    final @NotNull Hint finalHint = hint;
     final @Nullable TraceContext finalTraceContext = traceContext;
-    if (options.isEnableAsyncProcessing()) {
+    final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(hint);
+    if (shouldProcessAsync(hint)) {
+      final @NotNull Hint asyncHint = hint;
+      final @Nullable IScope asyncScope = snapshotScope(scope);
       if (!asyncEventProcessingExecutor.submit(
           () ->
               captureTransactionAfterProcessing(
                   finalTransaction,
                   finalTraceContext,
-                  scope,
-                  finalHint,
+                  asyncScope,
+                  asyncHint,
                   profilingTraceData,
-                  HintUtils.shouldApplyScopeData(finalHint)))) {
+                  applyScopedProcessors),
+          () -> recordLostTransaction(DiscardReason.QUEUE_OVERFLOW, finalTransaction))) {
         recordLostTransaction(DiscardReason.QUEUE_OVERFLOW, transaction);
         return SentryId.EMPTY_ID;
       }
@@ -1345,12 +1398,7 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
     }
 
     return captureTransactionAfterProcessing(
-        transaction,
-        traceContext,
-        scope,
-        hint,
-        profilingTraceData,
-        HintUtils.shouldApplyScopeData(hint));
+        transaction, traceContext, scope, hint, profilingTraceData, applyScopedProcessors);
   }
 
   private @NotNull SentryId captureTransactionAfterProcessing(
@@ -1583,13 +1631,18 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
 
     final SentryId sentryId = event.getEventId() != null ? event.getEventId() : SentryId.EMPTY_ID;
     final @NotNull SentryEvent finalEvent = event;
-    final @NotNull Hint finalHint = hint;
-    if (options.isEnableAsyncProcessing()) {
-      final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(finalHint);
+    final boolean applyScopedProcessors = HintUtils.shouldApplyScopeData(hint);
+    if (shouldProcessAsync(hint)) {
+      final @NotNull Hint asyncHint = hint;
+      final @NotNull IScope asyncScope = scope.clone();
       if (!asyncEventProcessingExecutor.submit(
           () ->
               captureFeedbackAfterProcessing(
-                  finalEvent, feedback, finalHint, scope, applyScopedProcessors))) {
+                  finalEvent, feedback, asyncHint, asyncScope, applyScopedProcessors),
+          () ->
+              options
+                  .getClientReportRecorder()
+                  .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Feedback))) {
         options
             .getClientReportRecorder()
             .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.Feedback);
@@ -1598,8 +1651,7 @@ public void captureSession(final @NotNull Session session, final @Nullable Hint
       return sentryId;
     }
 
-    return captureFeedbackAfterProcessing(
-        event, feedback, hint, scope, HintUtils.shouldApplyScopeData(hint));
+    return captureFeedbackAfterProcessing(event, feedback, hint, scope, applyScopedProcessors);
   }
 
   private @NotNull SentryId captureFeedbackAfterProcessing(
@@ -1718,8 +1770,10 @@ public void captureLog(@Nullable SentryLogEvent logEvent, @Nullable IScope scope
     if (logEvent != null) {
       final @NotNull SentryLogEvent finalLogEvent = logEvent;
       if (options.isEnableAsyncProcessing()) {
+        final @Nullable IScope asyncScope = snapshotScope(scope);
         if (!asyncEventProcessingExecutor.submit(
-            () -> captureLogAfterProcessing(finalLogEvent, scope))) {
+            () -> captureLogAfterProcessing(finalLogEvent, asyncScope),
+            () -> recordLostLog(DiscardReason.QUEUE_OVERFLOW, finalLogEvent))) {
           recordLostLog(DiscardReason.QUEUE_OVERFLOW, finalLogEvent);
         }
         return;
@@ -1795,10 +1849,15 @@ public void captureMetric(
 
     if (metricsEvent != null) {
       final @NotNull SentryMetricsEvent finalMetricsEvent = metricsEvent;
-      final @NotNull Hint finalHint = hint;
-      if (options.isEnableAsyncProcessing()) {
+      if (shouldProcessAsync(hint)) {
+        final @NotNull Hint asyncHint = hint;
+        final @Nullable IScope asyncScope = snapshotScope(scope);
         if (!asyncEventProcessingExecutor.submit(
-            () -> captureMetricAfterProcessing(finalMetricsEvent, scope, finalHint))) {
+            () -> captureMetricAfterProcessing(finalMetricsEvent, asyncScope, asyncHint),
+            () ->
+                options
+                    .getClientReportRecorder()
+                    .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.TraceMetric))) {
           options
               .getClientReportRecorder()
               .recordLostEvent(DiscardReason.QUEUE_OVERFLOW, DataCategory.TraceMetric);
@@ -2197,8 +2256,10 @@ public void close(final boolean isRestarting) {
     try {
       final long timeoutMillis = isRestarting ? 0 : options.getShutdownTimeoutMillis();
       final long deadline = System.currentTimeMillis() + timeoutMillis;
-      flush(timeoutMillis);
-      asyncEventProcessingExecutor.close(remainingTimeout(deadline));
+      // Quiesce the async stage before draining downstream: a task finishing after the transport
+      // flush would hand its envelope to a transport that is already closing.
+      asyncEventProcessingExecutor.close(timeoutMillis);
+      flush(remainingTimeout(deadline));
       loggerBatchProcessor.close(isRestarting);
       metricsBatchProcessor.close(isRestarting);
       transport.close(isRestarting);
diff --git a/sentry/src/test/java/io/sentry/AsyncEventProcessingExecutorTest.kt b/sentry/src/test/java/io/sentry/AsyncEventProcessingExecutorTest.kt
new file mode 100644
index 00000000000..7812ecc68f4
--- /dev/null
+++ b/sentry/src/test/java/io/sentry/AsyncEventProcessingExecutorTest.kt
@@ -0,0 +1,151 @@
+package io.sentry
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.system.measureNanoTime
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class AsyncEventProcessingExecutorTest {
+  private val noOpDrop = Runnable {}
+
+  private fun getSut(maxQueueSize: Int = 10) =
+    AsyncEventProcessingExecutor(maxQueueSize, NoOpLogger.getInstance())
+
+  @Test
+  fun `submit runs the task off the caller thread`() {
+    val sut = getSut()
+    val done = CountDownLatch(1)
+    var taskThread: String? = null
+
+    assertTrue(
+      sut.submit(
+        {
+          taskThread = Thread.currentThread().name
+          done.countDown()
+        },
+        noOpDrop,
+      )
+    )
+
+    assertTrue(done.await(5, TimeUnit.SECONDS))
+    assertTrue(taskThread!!.startsWith("SentryAsyncEventProcessing-"))
+    sut.close(5000)
+  }
+
+  @Test
+  fun `submit rejects once the queue is full`() {
+    val sut = getSut(maxQueueSize = 1)
+    val block = CountDownLatch(1)
+    val started = CountDownLatch(1)
+
+    assertTrue(
+      sut.submit(
+        {
+          started.countDown()
+          block.await(5, TimeUnit.SECONDS)
+        },
+        noOpDrop,
+      )
+    )
+    assertTrue(started.await(5, TimeUnit.SECONDS))
+    assertFalse(sut.submit({}, noOpDrop))
+
+    block.countDown()
+    sut.close(5000)
+  }
+
+  @Test
+  fun `submit rejects after close`() {
+    val sut = getSut()
+    sut.close(5000)
+
+    assertTrue(sut.isClosed)
+    assertFalse(sut.submit({}, noOpDrop))
+  }
+
+  @Test
+  fun `close reports tasks that never ran`() {
+    val sut = getSut()
+    val block = CountDownLatch(1)
+    val started = CountDownLatch(1)
+    val dropped = AtomicInteger()
+
+    sut.submit(
+      {
+        started.countDown()
+        block.await(10, TimeUnit.SECONDS)
+      },
+      noOpDrop,
+    )
+    assertTrue(started.await(5, TimeUnit.SECONDS))
+    // Queued behind the blocked worker, so shutdownNow never gets to run it.
+    sut.submit({}, { dropped.incrementAndGet() })
+
+    sut.close(100)
+
+    assertEquals(1, dropped.get())
+    block.countDown()
+  }
+
+  @Test
+  fun `submit does not block behind a draining close`() {
+    val sut = getSut()
+    val block = CountDownLatch(1)
+    val started = CountDownLatch(1)
+    sut.submit(
+      {
+        started.countDown()
+        block.await(10, TimeUnit.SECONDS)
+      },
+      noOpDrop,
+    )
+    assertTrue(started.await(5, TimeUnit.SECONDS))
+
+    val closing = Thread { sut.close(30000) }
+    closing.start()
+    awaitThreadState(closing, Thread.State.TIMED_WAITING)
+
+    // close() is parked in awaitTermination. Submitting must return right away (rejected, since
+    // the executor is closed) rather than block on a lock held for the whole shutdown.
+    val durationNanos = measureNanoTime { assertFalse(sut.submit({}, noOpDrop)) }
+
+    block.countDown()
+    closing.join(15000)
+    assertTrue(
+      durationNanos < TimeUnit.SECONDS.toNanos(5),
+      "submit blocked for ${TimeUnit.NANOSECONDS.toMillis(durationNanos)}ms during close",
+    )
+  }
+
+  @Test
+  fun `waitTillIdle returns once accepted work finishes`() {
+    val sut = getSut()
+    val block = CountDownLatch(1)
+    val finished = AtomicInteger()
+    sut.submit(
+      {
+        block.await(5, TimeUnit.SECONDS)
+        finished.incrementAndGet()
+      },
+      noOpDrop,
+    )
+
+    block.countDown()
+    sut.waitTillIdle(5000)
+
+    assertEquals(1, finished.get())
+    sut.close(5000)
+  }
+
+  private fun awaitThreadState(thread: Thread, state: Thread.State) {
+    val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5)
+    while (thread.state != state && System.nanoTime() < deadline) {
+      Thread.sleep(1)
+    }
+    assertEquals(state, thread.state)
+  }
+}
diff --git a/sentry/src/test/java/io/sentry/SentryClientTest.kt b/sentry/src/test/java/io/sentry/SentryClientTest.kt
index 7e80e0cadc7..7d8fc9fb242 100644
--- a/sentry/src/test/java/io/sentry/SentryClientTest.kt
+++ b/sentry/src/test/java/io/sentry/SentryClientTest.kt
@@ -48,6 +48,7 @@ import java.util.LinkedList
 import java.util.UUID
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFails
@@ -348,6 +349,138 @@ class SentryClientTest {
     sut.flush(5000)
   }
 
+  @Test
+  fun `close stops accepting async work before draining downstream`() {
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    fixture.sentryOptions.shutdownTimeoutMillis = 10000
+    // A capture that lands while the transport is being flushed. With the async stage quiesced
+    // first, it is rejected outright; otherwise its envelope reaches an already-drained transport.
+    val lateCapture = AtomicReference<SentryId>()
+    val client = AtomicReference<SentryClient>()
+    whenever(fixture.transport.flush(any())).doAnswer {
+      lateCapture.compareAndSet(null, client.get().captureEvent(SentryEvent()))
+      null
+    }
+    val sut = fixture.getSut()
+    client.set(sut)
+
+    sut.close(false)
+
+    assertEquals(SentryId.EMPTY_ID, lateCapture.get())
+    val inOrder = inOrder(fixture.transport)
+    inOrder.verify(fixture.transport).flush(any())
+    inOrder.verify(fixture.transport).close(eq(false))
+  }
+
+  @Test
+  fun `close records lost events for async work that could not be drained`() {
+    val continueAsync = CountDownLatch(1)
+    val asyncStarted = CountDownLatch(1)
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    fixture.sentryOptions.shutdownTimeoutMillis = 100
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          asyncStarted.countDown()
+          assertTrue(continueAsync.await(10, TimeUnit.SECONDS))
+          return event
+        }
+      }
+    )
+    val sut = fixture.getSut()
+    // The first event occupies the single worker thread, the second one is stuck in the queue and
+    // is the one that gets dropped by shutdownNow.
+    sut.captureEvent(SentryEvent())
+    assertTrue(asyncStarted.await(5, TimeUnit.SECONDS))
+    sut.captureEvent(SentryEvent())
+
+    sut.close(false)
+
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.QUEUE_OVERFLOW.reason, DataCategory.Error.category, 1)),
+    )
+    continueAsync.countDown()
+  }
+
+  @Test
+  fun `crash hints bypass the async queue and send on the caller thread`() {
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    val processingThread = AtomicReference<String>()
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          processingThread.set(Thread.currentThread().name)
+          return event
+        }
+      }
+    )
+    val sut = fixture.getSut()
+    val hint = HintUtils.createWithTypeCheckHint(DiskFlushNotificationHint())
+
+    sut.captureEvent(SentryEvent(), hint)
+
+    // Returning means the event is already sent - no flush needed.
+    verify(fixture.transport).send(any(), anyOrNull())
+    assertEquals(Thread.currentThread().name, processingThread.get())
+  }
+
+  @Test
+  fun `cached and backfillable hints bypass the async queue`() {
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    val sut = fixture.getSut()
+
+    sut.captureEvent(SentryEvent(), HintUtils.createWithTypeCheckHint(CachedHint()))
+    sut.captureEvent(SentryEvent(), HintUtils.createWithTypeCheckHint(BackfillableHint()))
+
+    verify(fixture.transport, times(2)).send(any(), anyOrNull())
+  }
+
+  @Test
+  fun `async processing sees the scope as of the capture call`() {
+    fixture.sentryOptions.isEnableAsyncProcessing = true
+    val continueAsync = CountDownLatch(1)
+    val submitted = CountDownLatch(1)
+    val observedTag = AtomicReference<String>()
+    fixture.sentryOptions.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          submitted.countDown()
+          assertTrue(continueAsync.await(5, TimeUnit.SECONDS))
+          return event
+        }
+      }
+    )
+    val sut = fixture.getSut()
+    val scope = Scope(fixture.sentryOptions)
+    scope.setTag("state", "at-capture")
+    scope.addEventProcessor(
+      object : EventProcessor {
+        override fun processAsync(event: SentryEvent, hint: Hint): SentryEvent? {
+          observedTag.set(event.tags?.get("state"))
+          return event
+        }
+      }
+    )
+
+    sut.captureEvent(SentryEvent(), scope, null)
+    assertTrue(submitted.await(5, TimeUnit.SECONDS))
+    // Caller moves on while the async stage is still queued.
+    scope.setTag("state", "after-capture")
+    continueAsync.countDown()
+    sut.flush(5000)
+
+    assertEquals("at-capture", observedTag.get())
+  }
+
+  private class DiskFlushNotificationHint : DiskFlushNotification {
+    override fun markFlushed() = Unit
+
+    override fun isFlushable(eventId: SentryId?): Boolean = true
+
+    override fun setFlushable(eventId: SentryId) = Unit
+  }
+
   @Test
   fun `when beforeSend is returns null, event is dropped`() {
     fixture.sentryOptions.setBeforeSend { _: SentryEvent, _: Any? -> null }