diff --git a/CHANGELOG.md b/CHANGELOG.md
index eaed5ce3eb8..246668e0744 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -124,6 +124,7 @@
 
 - Fix SDK callback error handling ([#6140](https://github.com/getsentry/sentry-java/pull/6140))
   - Add `DiscardReason.CALLBACK_ERROR` and use it for telemetry dropped when a `beforeSend*` callback throws. `OnDiscardCallback` can now receive this value.
+  - Drop telemetry and record `callback_error` when a customer event processor throws instead of continuing with a potentially partially processed item. SDK-owned processor failures are logged and processing continues without a `callback_error` client report.
 - Disable URL caching when reading `META-INF/MANIFEST.MF` files during version detection so that the SDK no longer keeps jar file handles open for the life of the process ([#6124](https://github.com/getsentry/sentry-java/pull/6124)
 - Keep the `EventListener` wrapped by `SentryOkHttpEventListener` per `Call` ([#6003](https://github.com/getsentry/sentry-java/pull/6003))
 
diff --git a/sentry/src/main/java/io/sentry/SentryClient.java b/sentry/src/main/java/io/sentry/SentryClient.java
index a9cc392e711..705f6fe3bbb 100644
--- a/sentry/src/main/java/io/sentry/SentryClient.java
+++ b/sentry/src/main/java/io/sentry/SentryClient.java
@@ -8,6 +8,7 @@
 import io.sentry.hints.Cached;
 import io.sentry.hints.DiskFlushNotification;
 import io.sentry.hints.TransactionEnd;
+import io.sentry.internal.eventprocessor.SentryEventProcessor;
 import io.sentry.logger.ILoggerBatchProcessor;
 import io.sentry.logger.NoOpLoggerBatchProcessor;
 import io.sentry.metrics.IMetricsBatchProcessor;
@@ -506,6 +507,12 @@ private SentryEvent processEvent(
                 e,
                 "An exception occurred while processing event by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(DiscardReason.CALLBACK_ERROR, DataCategory.Error);
+          return null;
+        }
       }
 
       if (event == null) {
@@ -557,6 +564,10 @@ private SentryLogEvent processLogEvent(
                 e,
                 "An exception occurred while processing log event by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          recordLostLogEvent(DiscardReason.CALLBACK_ERROR, eventBeforeProcessor);
+          return null;
+        }
       }
 
       if (event == null) {
@@ -590,6 +601,10 @@ private SentryMetricsEvent processMetricsEvent(
                 e,
                 "An exception occurred while processing metrics event by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          recordLostMetricsEvent(DiscardReason.CALLBACK_ERROR, eventBeforeProcessor);
+          return null;
+        }
       }
 
       if (event == null) {
@@ -622,6 +637,16 @@ private SentryMetricsEvent processMetricsEvent(
                 e,
                 "An exception occurred while processing transaction by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(DiscardReason.CALLBACK_ERROR, DataCategory.Transaction);
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(
+                  DiscardReason.CALLBACK_ERROR, DataCategory.Span, spanCountBeforeProcessor + 1);
+          return null;
+        }
       }
       final int spanCountAfterProcessor = transaction == null ? 0 : transaction.getSpans().size();
 
@@ -675,6 +700,12 @@ private SentryReplayEvent processReplayEvent(
                 e,
                 "An exception occurred while processing replay event by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(DiscardReason.CALLBACK_ERROR, DataCategory.Replay);
+          return null;
+        }
       }
 
       if (replayEvent == null) {
@@ -709,6 +740,12 @@ private SentryEvent processFeedbackEvent(
                 e,
                 "An exception occurred while processing feedback event by processor: %s",
                 processor.getClass().getName());
+        if (!(processor instanceof SentryEventProcessor)) {
+          options
+              .getClientReportRecorder()
+              .recordLostEvent(DiscardReason.CALLBACK_ERROR, DataCategory.Feedback);
+          return null;
+        }
       }
 
       if (feedbackEvent == null) {
diff --git a/sentry/src/test/java/io/sentry/SentryClientInternalEventProcessorTest.kt b/sentry/src/test/java/io/sentry/SentryClientInternalEventProcessorTest.kt
new file mode 100644
index 00000000000..59aaaa0a1ae
--- /dev/null
+++ b/sentry/src/test/java/io/sentry/SentryClientInternalEventProcessorTest.kt
@@ -0,0 +1,229 @@
+package io.sentry
+
+import com.google.common.truth.Truth.assertThat
+import io.sentry.clientreport.ClientReportTestHelper.Companion.assertClientReport
+import io.sentry.clientreport.DiscardReason
+import io.sentry.clientreport.DiscardedEvent
+import io.sentry.internal.eventprocessor.SentryEventProcessor
+import io.sentry.protocol.Feedback
+import io.sentry.protocol.SentryId
+import io.sentry.protocol.SentryTransaction
+import kotlin.test.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.mockito.kotlin.any
+import org.mockito.kotlin.anyOrNull
+import org.mockito.kotlin.check
+import org.mockito.kotlin.doAnswer
+import org.mockito.kotlin.eq
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.same
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.verifyNoInteractions
+import org.mockito.kotlin.whenever
+
+@RunWith(Parameterized::class)
+class SentryClientInternalEventProcessorTest(private val onScope: Boolean) {
+  companion object {
+    @JvmStatic
+    @Parameterized.Parameters(name = "onScope={0}")
+    fun data(): List<Array<Boolean>> = listOf(arrayOf(false), arrayOf(true))
+  }
+
+  private val fixture = SentryClientTest.Fixture()
+  private val options = fixture.sentryOptions
+  private val scope = Scope(options)
+  private val processor = mock<SentryEventProcessor>()
+  private val nextProcessor = mock<EventProcessor>()
+  private val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+  private val logger = mock<ILogger>()
+  private val failure = IllegalStateException("SDK processor failed")
+
+  init {
+    options.eventProcessors.clear()
+    options.onDiscard = onDiscard
+    options.setLogger(logger)
+    options.logs.isEnabled = true
+    options.metrics.isEnabled = true
+    if (onScope) {
+      scope.addEventProcessor(processor)
+      scope.addEventProcessor(nextProcessor)
+    } else {
+      options.addEventProcessor(processor)
+      options.addEventProcessor(nextProcessor)
+    }
+  }
+
+  @Test
+  fun `SDK event processor failure keeps event and runs remaining callbacks`() {
+    val event = SentryEvent()
+    val beforeSend = mock<SentryOptions.BeforeSendCallback>()
+    whenever(processor.process(any<SentryEvent>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryEvent>(), any())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any(), any())).thenAnswer { it.arguments[0] }
+    options.beforeSend = beforeSend
+
+    val id = fixture.getSut().captureEvent(event, scope)
+
+    assertThat(id).isEqualTo(event.eventId)
+    verify(nextProcessor).process(same(event), any())
+    verify(beforeSend).execute(same(event), any())
+    verify(fixture.transport)
+      .send(check { assertThat(it.header.eventId).isEqualTo(id) }, anyOrNull())
+    assertFailureLoggedWithoutLoss("event")
+  }
+
+  @Test
+  fun `SDK transaction processor failure keeps transaction and spans`() {
+    val transaction = SentryTransaction(fixture.sentryTracer)
+    val beforeSend = mock<SentryOptions.BeforeSendTransactionCallback>()
+    whenever(processor.process(any<SentryTransaction>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryTransaction>(), any())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any(), any())).thenAnswer { it.arguments[0] }
+    options.beforeSendTransaction = beforeSend
+
+    val id = fixture.getSut().captureTransaction(transaction, scope, null)
+
+    assertThat(id).isEqualTo(transaction.eventId)
+    verify(nextProcessor).process(same(transaction), any())
+    verify(beforeSend).execute(same(transaction), any())
+    verify(fixture.transport)
+      .send(
+        check {
+          val sent = it.items.first().getTransaction(options.serializer)!!
+          assertThat(sent.eventId).isEqualTo(id)
+          assertThat(sent.spans).hasSize(1)
+        },
+        anyOrNull(),
+      )
+    assertFailureLoggedWithoutLoss("transaction")
+  }
+
+  @Test
+  fun `SDK feedback processor failure keeps feedback and runs remaining callbacks`() {
+    val feedback = Feedback("message")
+    val beforeSend = mock<SentryOptions.BeforeSendCallback>()
+    whenever(processor.process(any<SentryEvent>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryEvent>(), any())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any(), any())).thenAnswer { it.arguments[0] }
+    options.beforeSendFeedback = beforeSend
+
+    val id = fixture.getSut().captureFeedback(feedback, null, scope)
+
+    assertThat(id).isNotEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor)
+      .process(
+        check<SentryEvent> {
+          assertThat(it.contexts.feedback).isSameInstanceAs(feedback)
+        },
+        any(),
+      )
+    verify(beforeSend).execute(check { assertThat(it.eventId).isEqualTo(id) }, any())
+    verify(fixture.transport)
+      .send(check { assertThat(it.header.eventId).isEqualTo(id) }, anyOrNull())
+    assertFailureLoggedWithoutLoss("feedback event")
+  }
+
+  @Test
+  fun `SDK log processor failure keeps log and runs remaining callbacks`() {
+    val event = SentryLogEvent(SentryId(), SentryNanotimeDate(), "message", SentryLogLevel.WARN)
+    val beforeSend = mock<SentryOptions.Logs.BeforeSendLogCallback>()
+    whenever(processor.process(any<SentryLogEvent>())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryLogEvent>())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any())).thenAnswer { it.arguments[0] }
+    options.logs.beforeSend = beforeSend
+
+    fixture.getSut().captureLog(event, scope)
+
+    verify(nextProcessor).process(same(event))
+    verify(beforeSend).execute(same(event))
+    verify(fixture.loggerBatchProcessor).add(same(event))
+    assertFailureLoggedWithoutLoss("log event")
+  }
+
+  @Test
+  fun `SDK metric processor failure keeps metric and runs remaining callbacks`() {
+    val event = SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "name", "gauge", 123.0)
+    val beforeSend = mock<SentryOptions.Metrics.BeforeSendMetricCallback>()
+    whenever(processor.process(any<SentryMetricsEvent>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryMetricsEvent>(), any())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any(), any())).thenAnswer { it.arguments[0] }
+    options.metrics.beforeSend = beforeSend
+
+    fixture.getSut().captureMetric(event, scope, null)
+
+    verify(nextProcessor).process(same(event), any())
+    verify(beforeSend).execute(same(event), any())
+    verify(fixture.metricsBatchProcessor).add(same(event))
+    assertFailureLoggedWithoutLoss("metrics event")
+  }
+
+  @Test
+  fun `SDK processor returning null still drops event as event_processor`() {
+    whenever(processor.process(any<SentryEvent>(), any())).thenReturn(null)
+
+    val id = fixture.getSut().captureEvent(SentryEvent(), scope)
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor, never()).process(any<SentryEvent>(), any())
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      options.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.EVENT_PROCESSOR.reason, DataCategory.Error.category, 1)),
+    )
+  }
+
+  @Test
+  fun `customer processor failure after SDK processor failure still drops event`() {
+    whenever(processor.process(any<SentryEvent>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryEvent>(), any()))
+      .thenThrow(IllegalArgumentException("customer"))
+    val beforeSend = mock<SentryOptions.BeforeSendCallback>()
+    options.beforeSend = beforeSend
+
+    val id = fixture.getSut().captureEvent(SentryEvent(), scope)
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor).process(any<SentryEvent>(), any())
+    verifyNoInteractions(beforeSend)
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      options.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Error.category, 1)),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Error, 1)
+  }
+
+  @Test
+  fun `spans removed before SDK processor failure retain event_processor accounting`() {
+    val transaction = SentryTransaction(fixture.sentryTracer)
+    whenever(processor.process(any<SentryTransaction>(), any())).doAnswer {
+      transaction.spans.clear()
+      throw failure
+    }
+    whenever(nextProcessor.process(any<SentryTransaction>(), any())).thenAnswer { it.arguments[0] }
+
+    val id = fixture.getSut().captureTransaction(transaction, scope, null)
+
+    assertThat(id).isEqualTo(transaction.eventId)
+    verify(nextProcessor).process(same(transaction), any())
+    verify(fixture.transport).send(any(), anyOrNull())
+    assertClientReport(
+      options.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.EVENT_PROCESSOR.reason, DataCategory.Span.category, 1)),
+    )
+  }
+
+  private fun assertFailureLoggedWithoutLoss(item: String) {
+    verify(logger)
+      .log(
+        eq(SentryLevel.ERROR),
+        same(failure),
+        eq("An exception occurred while processing $item by processor: %s"),
+        eq(processor.javaClass.name),
+      )
+    assertClientReport(options.clientReportRecorder, emptyList())
+    verifyNoInteractions(onDiscard)
+  }
+}
diff --git a/sentry/src/test/java/io/sentry/SentryClientTest.kt b/sentry/src/test/java/io/sentry/SentryClientTest.kt
index 7b907df1e82..99d12e12184 100644
--- a/sentry/src/test/java/io/sentry/SentryClientTest.kt
+++ b/sentry/src/test/java/io/sentry/SentryClientTest.kt
@@ -1,5 +1,6 @@
 package io.sentry
 
+import com.google.common.truth.Truth.assertThat
 import io.sentry.Scope.IWithPropagationContext
 import io.sentry.SentryLevel.WARNING
 import io.sentry.Session.State.Crashed
@@ -13,6 +14,7 @@ import io.sentry.hints.Backfillable
 import io.sentry.hints.Cached
 import io.sentry.hints.DiskFlushNotification
 import io.sentry.hints.TransactionEnd
+import io.sentry.internal.eventprocessor.SentryEventProcessor
 import io.sentry.logger.ILoggerBatchProcessor
 import io.sentry.logger.ILoggerBatchProcessorFactory
 import io.sentry.metrics.IMetricsBatchProcessor
@@ -474,6 +476,48 @@ class SentryClientTest {
     )
   }
 
+  @Test
+  fun `throwing log processor drops log and stops callbacks`() {
+    val scope = createScope()
+    val logEvent = SentryLogEvent(SentryId(), SentryNanotimeDate(), "message", SentryLogLevel.WARN)
+    val logEventNumberOfBytes =
+      JsonSerializationUtils.byteSizeOf(
+        fixture.sentryOptions.serializer,
+        fixture.sentryOptions.logger,
+        logEvent,
+      )
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.Logs.BeforeSendLogCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryLogEvent>()))
+      .thenThrow(IllegalStateException("test"))
+    scope.addEventProcessor(throwingProcessor)
+    scope.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.logs.beforeSend = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    fixture.getSut().captureLog(logEvent, scope)
+
+    verify(nextProcessor, never()).process(any<SentryLogEvent>())
+    verify(beforeSend, never()).execute(any())
+    verify(fixture.loggerBatchProcessor, never()).add(any())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(
+        DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.LogItem.category, 1),
+        DiscardedEvent(
+          DiscardReason.CALLBACK_ERROR.reason,
+          DataCategory.LogByte.category,
+          logEventNumberOfBytes,
+        ),
+      ),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.LogItem, 1)
+    verify(onDiscard)
+      .execute(DiscardReason.CALLBACK_ERROR, DataCategory.LogByte, logEventNumberOfBytes)
+  }
+
   @Test
   fun `when beforeSendLog is returns new instance, new instance is sent`() {
     val scope = createScope()
@@ -595,6 +639,56 @@ class SentryClientTest {
     )
   }
 
+  @Test
+  fun `throwing metric processor drops metric and stops callbacks`() {
+    val scope = createScope()
+    val metricsEvent = SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "name", "gauge", 123.0)
+    val metricsEventNumberOfBytes =
+      JsonSerializationUtils.byteSizeOf(
+        fixture.sentryOptions.serializer,
+        fixture.sentryOptions.logger,
+        metricsEvent,
+      )
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.Metrics.BeforeSendMetricCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryMetricsEvent>(), anyOrNull()))
+      .thenThrow(IllegalStateException("test"))
+    scope.addEventProcessor(throwingProcessor)
+    scope.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.metrics.beforeSend = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    fixture.getSut().captureMetric(metricsEvent, scope, null)
+
+    verify(nextProcessor, never()).process(any<SentryMetricsEvent>(), anyOrNull())
+    verify(beforeSend, never()).execute(any(), anyOrNull())
+    verify(fixture.metricsBatchProcessor, never()).add(any())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(
+        DiscardedEvent(
+          DiscardReason.CALLBACK_ERROR.reason,
+          DataCategory.TraceMetric.category,
+          1,
+        ),
+        DiscardedEvent(
+          DiscardReason.CALLBACK_ERROR.reason,
+          DataCategory.TraceMetricByte.category,
+          metricsEventNumberOfBytes,
+        ),
+      ),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.TraceMetric, 1)
+    verify(onDiscard)
+      .execute(
+        DiscardReason.CALLBACK_ERROR,
+        DataCategory.TraceMetricByte,
+        metricsEventNumberOfBytes,
+      )
+  }
+
   @Test
   fun `when beforeSendMetric is returns new instance, new instance is sent`() {
     val scope = createScope()
@@ -1242,6 +1336,42 @@ class SentryClientTest {
     )
   }
 
+  @Test
+  fun `throwing transaction processor drops transaction and stops callbacks`() {
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.BeforeSendTransactionCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryTransaction>(), anyOrNull()))
+      .thenThrow(IllegalStateException("test"))
+    fixture.sentryOptions.addEventProcessor(throwingProcessor)
+    fixture.sentryOptions.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.beforeSendTransaction = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    val id =
+      fixture
+        .getSut()
+        .captureTransaction(
+          SentryTransaction(fixture.sentryTracer),
+          fixture.sentryTracer.traceContext(),
+        )
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor, never()).process(any<SentryTransaction>(), anyOrNull())
+    verify(beforeSend, never()).execute(any(), anyOrNull())
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(
+        DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Transaction.category, 1),
+        DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Span.category, 2),
+      ),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Transaction, 1)
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Span, 2)
+  }
+
   @Test
   fun `transaction dropped by ignoredTransactions is recorded`() {
     fixture.sentryOptions.setIgnoredTransactions(listOf("a-transaction"))
@@ -1927,10 +2057,29 @@ class SentryClientTest {
   }
 
   @Test
-  fun `exception thrown by an event processor is handled gracefully`() {
-    fixture.sentryOptions.addEventProcessor(eventProcessorThrows())
-    val sut = fixture.getSut()
-    sut.captureEvent(SentryEvent())
+  fun `exception thrown by an event processor drops event and stops callbacks`() {
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.BeforeSendCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryEvent>(), anyOrNull()))
+      .thenThrow(IllegalStateException("test"))
+    fixture.sentryOptions.addEventProcessor(throwingProcessor)
+    fixture.sentryOptions.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.beforeSend = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    val id = fixture.getSut().captureEvent(SentryEvent())
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor, never()).process(any<SentryEvent>(), anyOrNull())
+    verify(beforeSend, never()).execute(any(), anyOrNull())
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Error.category, 1)),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Error, 1)
   }
 
   @Test
@@ -3524,6 +3673,74 @@ class SentryClientTest {
     verify(onDiscardMock, times(1)).execute(DiscardReason.EVENT_PROCESSOR, DataCategory.Replay, 1)
   }
 
+  @Test
+  fun `throwing replay processor drops replay and stops callbacks`() {
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.BeforeSendReplayCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryReplayEvent>(), anyOrNull()))
+      .thenThrow(IllegalStateException("test"))
+    fixture.sentryOptions.addEventProcessor(throwingProcessor)
+    fixture.sentryOptions.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.beforeSendReplay = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    val id = fixture.getSut().captureReplayEvent(createReplayEvent(), createScope(), null)
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor, never()).process(any<SentryReplayEvent>(), anyOrNull())
+    verify(beforeSend, never()).execute(any(), anyOrNull())
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Replay.category, 1)),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Replay, 1)
+  }
+
+  @Test
+  fun `throwing SDK replay processor keeps replay and runs remaining callbacks`() {
+    val processor = mock<SentryEventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.BeforeSendReplayCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    val logger = mock<ILogger>()
+    val failure = IllegalStateException("SDK processor failed")
+    val replay = createReplayEvent()
+    whenever(processor.process(any<SentryReplayEvent>(), any())).thenThrow(failure)
+    whenever(nextProcessor.process(any<SentryReplayEvent>(), any())).thenAnswer { it.arguments[0] }
+    whenever(beforeSend.execute(any(), any())).thenAnswer { it.arguments[0] }
+    fixture.sentryOptions.addEventProcessor(processor)
+    fixture.sentryOptions.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.beforeSendReplay = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+    fixture.sentryOptions.setLogger(logger)
+
+    val id = fixture.getSut().captureReplayEvent(replay, createScope(), null)
+
+    assertThat(id).isEqualTo(replay.eventId)
+    verify(nextProcessor).process(eq(replay), any())
+    verify(beforeSend).execute(eq(replay), any())
+    verify(fixture.transport)
+      .send(
+        check {
+          assertThat(it.header.eventId).isEqualTo(id)
+          assertThat(it.items.first().header.type).isEqualTo(SentryItemType.ReplayVideo)
+        },
+        anyOrNull(),
+      )
+    verify(logger)
+      .log(
+        eq(SentryLevel.ERROR),
+        eq(failure),
+        eq("An exception occurred while processing replay event by processor: %s"),
+        eq(processor.javaClass.name),
+      )
+    assertClientReport(fixture.sentryOptions.clientReportRecorder, emptyList())
+    verifyNoInteractions(onDiscard)
+  }
+
   @Test
   fun `calls captureReplay on replay controller for error events`() {
     var called = false
@@ -4086,6 +4303,34 @@ class SentryClientTest {
     verify(onDiscardMock, times(1)).execute(DiscardReason.EVENT_PROCESSOR, DataCategory.Feedback, 1)
   }
 
+  @Test
+  fun `throwing feedback processor drops feedback and stops callbacks`() {
+    val throwingProcessor = mock<EventProcessor>()
+    val nextProcessor = mock<EventProcessor>()
+    val beforeSend = mock<SentryOptions.BeforeSendCallback>()
+    val onDiscard = mock<SentryOptions.OnDiscardCallback>()
+    whenever(throwingProcessor.process(any<SentryEvent>(), anyOrNull()))
+      .thenThrow(IllegalStateException("test"))
+    fixture.sentryOptions.addEventProcessor(throwingProcessor)
+    fixture.sentryOptions.addEventProcessor(nextProcessor)
+    fixture.sentryOptions.beforeSendFeedback = beforeSend
+    fixture.sentryOptions.onDiscard = onDiscard
+
+    val id = fixture.getSut().captureFeedback(Feedback("message"), null, createScope())
+
+    assertThat(id).isEqualTo(SentryId.EMPTY_ID)
+    verify(nextProcessor, never()).process(any<SentryEvent>(), anyOrNull())
+    verify(beforeSend, never()).execute(any(), anyOrNull())
+    verify(fixture.transport, never()).send(any(), anyOrNull())
+    assertClientReport(
+      fixture.sentryOptions.clientReportRecorder,
+      listOf(
+        DiscardedEvent(DiscardReason.CALLBACK_ERROR.reason, DataCategory.Feedback.category, 1)
+      ),
+    )
+    verify(onDiscard).execute(DiscardReason.CALLBACK_ERROR, DataCategory.Feedback, 1)
+  }
+
   // endregion
 
   private fun givenScopeWithStartedSession(
@@ -4352,14 +4597,6 @@ class SentryClientTest {
     override fun timestamp(): Long? = null
   }
 
-  private fun eventProcessorThrows(): EventProcessor {
-    return object : EventProcessor {
-      override fun process(event: SentryEvent, hint: Hint): SentryEvent? {
-        throw Throwable()
-      }
-    }
-  }
-
   private class BackfillableHint : Backfillable {
     override fun shouldEnrich(): Boolean = false
   }