fix: guard executor shutdown in BaseCaptureStrategy.stop()

Each start/stop cycle leaked one SentryReplayPersister-* thread because
stop() reset delegated properties (segmentTimestamp, currentReplayId)
whose setters dispatch to persistingExecutor, initialising the lazy —
but stop() never shut it down.

Replace the lazy delegate with an explicit nullable holder so the
executor is only created when actually needed and can be detected at
stop() time.  Call shutdownNow() (non-blocking) rather than the blocking
shutdown() to avoid ANRs when stop() runs on the main thread.

Fixes #5564

Author: tsushanth

sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt

@@ -58,11 +58,20 @@ internal abstract class BaseCaptureStrategy(
     private const val MAX_TRACE_IDS = 100
   }
 
-  private val persistingExecutor: ScheduledExecutorService by lazy {
-    val delegate =
-      Executors.newSingleThreadScheduledExecutor(ReplayPersistingExecutorServiceThreadFactory())
-    ReplayExecutorService(delegate, options)
-  }
+  // Explicit holder so we can detect whether the executor was ever initialised and shut it down
+  // without initialising it as a side-effect (which is what a plain `lazy` delegate would do).
+  private var persistingExecutorHolder: ScheduledExecutorService? = null
+  private val persistingExecutor: ScheduledExecutorService
+    get() {
+      return persistingExecutorHolder
+        ?: run {
+          val delegate =
+            Executors.newSingleThreadScheduledExecutor(
+              ReplayPersistingExecutorServiceThreadFactory()
+            )
+          ReplayExecutorService(delegate, options).also { persistingExecutorHolder = it }
+        }
+    }
   private val gestureConverter = ReplayGestureConverter(dateProvider)
 
   protected val isTerminating = AtomicBoolean(false)
@@ -123,6 +132,13 @@ internal abstract class BaseCaptureStrategy(
     replayStartTimestamp.set(0)
     segmentTimestamp = null
     currentReplayId = SentryId.EMPTY_ID
+    // Shut down the persisting executor only if it was actually initialised.  We must not call the
+    // blocking ReplayExecutorService.shutdown() here because stop() may run on the main thread,
+    // and blocking there risks an ANR.  shutdownNow() is non-blocking: it cancels queued tasks and
+    // interrupts any running one, which is acceptable because the tasks are best-effort persistence
+    // writes and the cache has already been closed above.
+    persistingExecutorHolder?.shutdownNow()
+    persistingExecutorHolder = null
   }
 
   protected fun createSegmentInternal(

sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt

@@ -7,6 +7,7 @@ import io.sentry.IScopes
 import io.sentry.Scope
 import io.sentry.ScopeCallback
 import io.sentry.SentryOptions
+import io.sentry.util.thread.IThreadChecker
 import io.sentry.SentryReplayEvent
 import io.sentry.SentryReplayEvent.ReplayType
 import io.sentry.SentryReplayOptions.SentryReplayQuality.HIGH
@@ -554,4 +555,35 @@ class SessionCaptureStrategyTest {
         any(),
       )
   }
+
+  @Test
+  fun `stop shuts down persisting executor so no SentryReplayPersister threads leak across cycles`() {
+    // Force isMainThread() to return true so that property-reset writes in stop() are dispatched
+    // to the persistingExecutor, which is the code path that used to leak threads.
+    fixture.options.threadChecker =
+      mock<IThreadChecker> {
+        on { isMainThread() }.thenReturn(true)
+        on { isMainThread(any<Thread>()) }.thenReturn(true)
+        on { isMainThread(anyLong()) }.thenReturn(false)
+        on { currentThreadName }.thenReturn("main")
+        on { currentThreadSystemId() }.thenReturn(0L)
+      }
+
+    repeat(3) {
+      val strategy = fixture.getSut()
+      strategy.start(0, SentryId())
+      strategy.onConfigurationChanged(fixture.recorderConfig)
+      strategy.stop()
+    }
+
+    // Allow time for thread termination after shutdownNow().
+    Thread.sleep(200)
+
+    val leakedThreads =
+      Thread.getAllStackTraces().keys.filter { it.name.startsWith("SentryReplayPersister-") }
+    assertTrue(
+      leakedThreads.isEmpty(),
+      "Expected no SentryReplayPersister threads after stop(), found: ${leakedThreads.map { it.name }}",
+    )
+  }
 }