RUM-18168: Fix DataFlusher racing the upload scheduler and duplicating events

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/CoreFeature.kt

@@ -359,7 +359,7 @@ internal class CoreFeature(
         contextExecutorService.queue.drainTo(contextTasks)
 
         contextExecutorService.shutdown()
-        contextExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
+        awaitTerminationLogged(contextExecutorService, "contextExecutorService", DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
         contextTasks.forEach {
             it.run()
         }
@@ -376,14 +376,38 @@ internal class CoreFeature(
         persistenceExecutorService.shutdown()
         uploadExecutorService.shutdown()
 
-        persistenceExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
-        uploadExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
+        awaitTerminationLogged(
+            persistenceExecutorService,
+            "persistenceExecutorService",
+            DRAIN_WAIT_SECONDS,
+            TimeUnit.SECONDS
+        )
+        // uploadExecutorService can be mid-upload when this runs, and only NETWORK_TIMEOUT_MS
+        // bounds how long that upload can take. Failing to wait long enough here can lead to
+        // a DataFlusher race where it uploads the same batch twice. See RUM-18168.
+        awaitTerminationLogged(
+            uploadExecutorService,
+            "uploadExecutorService",
+            NETWORK_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS
+        )
 
         ioTasks.forEach {
             it.run()
         }
     }
 
+    @Suppress("UnsafeThirdPartyFunctionCall") // Used in Nightly tests only
+    private fun awaitTerminationLogged(executor: ExecutorService, executorName: String, wait: Long, unit: TimeUnit) {
+        if (!executor.awaitTermination(wait, unit)) {
+            internalLogger.log(
+                InternalLogger.Level.WARN,
+                InternalLogger.Target.MAINTAINER,
+                { "drainAndShutdownExecutors: $executorName did not terminate within $wait $unit" }
+            )
+        }
+    }
+
     // region Internal
 
     @WorkerThread

dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/CoreFeatureTest.kt

@@ -1555,10 +1555,39 @@ internal class CoreFeatureTest {
         // Then
         inOrder(mockUploadService) {
             verify(mockUploadService).shutdown()
-            verify(mockUploadService).awaitTermination(10, TimeUnit.SECONDS)
+            verify(mockUploadService).awaitTermination(CoreFeature.NETWORK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
         }
     }
 
+    @Test
+    fun `M log a warning W drainAndShutdownExecutors() { upload executor doesn't terminate in time }`() {
+        // Given
+        testedFeature.initialize(
+            appContext.mockInstance,
+            fakeSdkInstanceId,
+            fakeConfig,
+            fakeConsent
+        )
+
+        val blockingQueue = LinkedBlockingQueue<Runnable>()
+        val mockUploadService: ScheduledThreadPoolExecutor = mock()
+        whenever(mockUploadService.queue).thenReturn(blockingQueue)
+        whenever(mockUploadService.awaitTermination(any(), any())) doReturn false
+        testedFeature.uploadExecutorService = mockUploadService
+
+        // When
+        testedFeature.drainAndShutdownExecutors()
+
+        // Then
+        mockInternalLogger.verifyLog(
+            InternalLogger.Level.WARN,
+            InternalLogger.Target.MAINTAINER,
+            "drainAndShutdownExecutors: uploadExecutorService did not terminate " +
+                "within ${CoreFeature.NETWORK_TIMEOUT_MS} ${TimeUnit.MILLISECONDS}",
+            mode = atLeastOnce()
+        )
+    }
+
     @Test
     fun `M shutdown with wait the context executor W drainAndShutdownExecutors()`() {
         // Given