diff --git a/dd-sdk-android/src/main/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnable.kt b/dd-sdk-android/src/main/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnable.kt
index a6f0d206f1..521befe2d7 100644
--- a/dd-sdk-android/src/main/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnable.kt
+++ b/dd-sdk-android/src/main/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnable.kt
@@ -32,7 +32,8 @@ internal class DataUploadRunnable(
     private val contextProvider: ContextProvider,
     private val networkInfoProvider: NetworkInfoProvider,
     private val systemInfoProvider: SystemInfoProvider,
-    uploadFrequency: UploadFrequency
+    uploadFrequency: UploadFrequency,
+    private val batchUploadWaitTimeoutMs: Long = CoreFeature.NETWORK_TIMEOUT_MS
 ) : UploadRunnable {
 
     internal var currentDelayIntervalMs = DEFAULT_DELAY_FACTOR * uploadFrequency.baseStepMs
@@ -55,18 +56,21 @@ internal class DataUploadRunnable(
                     lock.countDown()
                 }
             ) { batchId, reader ->
-                val batch = reader.read()
-                val batchMeta = reader.currentMetadata()
-
-                consumeBatch(
-                    context,
-                    batchId,
-                    batch,
-                    batchMeta
-                )
-                lock.countDown()
+                try {
+                    val batch = reader.read()
+                    val batchMeta = reader.currentMetadata()
+
+                    consumeBatch(
+                        context,
+                        batchId,
+                        batch,
+                        batchMeta
+                    )
+                } finally {
+                    lock.countDown()
+                }
             }
-            lock.await(CoreFeature.NETWORK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+            lock.await(batchUploadWaitTimeoutMs, TimeUnit.MILLISECONDS)
         }
 
         scheduleNextUpload()
diff --git a/dd-sdk-android/src/test/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnableTest.kt b/dd-sdk-android/src/test/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnableTest.kt
index 8eeee2acb1..cc34ae5414 100644
--- a/dd-sdk-android/src/test/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnableTest.kt
+++ b/dd-sdk-android/src/test/kotlin/com/datadog/android/v2/core/internal/data/upload/DataUploadRunnableTest.kt
@@ -20,10 +20,12 @@ import com.datadog.android.v2.core.internal.storage.BatchConfirmation
 import com.datadog.android.v2.core.internal.storage.BatchId
 import com.datadog.android.v2.core.internal.storage.BatchReader
 import com.datadog.android.v2.core.internal.storage.Storage
+import com.datadog.tools.unit.forge.aThrowable
 import com.nhaarman.mockitokotlin2.any
 import com.nhaarman.mockitokotlin2.argumentCaptor
 import com.nhaarman.mockitokotlin2.doAnswer
 import com.nhaarman.mockitokotlin2.doReturn
+import com.nhaarman.mockitokotlin2.doThrow
 import com.nhaarman.mockitokotlin2.eq
 import com.nhaarman.mockitokotlin2.mock
 import com.nhaarman.mockitokotlin2.same
@@ -113,7 +115,8 @@ internal class DataUploadRunnableTest {
             mockContextProvider,
             mockNetworkInfoProvider,
             mockSystemInfoProvider,
-            fakeUploadFrequency
+            fakeUploadFrequency,
+            TEST_BATCH_UPLOAD_WAIT_TIMEOUT_MS
         )
     }
 
@@ -328,7 +331,7 @@ internal class DataUploadRunnableTest {
     }
 
     @Test
-    fun `M not send batch W run() { batteryFullOrCharging, powerSaveMode}`(
+    fun `M not send batch W run() { batteryFullOrCharging, powerSaveMode }`(
         @IntForgery(min = 0, max = 100) batteryLevel: Int
     ) {
         // Given
@@ -353,7 +356,7 @@ internal class DataUploadRunnableTest {
     }
 
     @Test
-    fun `M not send batch W run() { batteryLeveHigh, powerSaveMode}`(
+    fun `M not send batch W run() { batteryLeveHigh, powerSaveMode }`(
         @IntForgery(min = DataUploadRunnable.LOW_BATTERY_THRESHOLD + 1) batteryLevel: Int
     ) {
         // Given
@@ -378,7 +381,7 @@ internal class DataUploadRunnableTest {
     }
 
     @Test
-    fun `M not send batch W run() { onExternalPower, powerSaveMode}`(
+    fun `M not send batch W run() { onExternalPower, powerSaveMode }`(
         @IntForgery(min = 0, max = DataUploadRunnable.LOW_BATTERY_THRESHOLD) batteryLevel: Int
     ) {
         // Given
@@ -880,4 +883,73 @@ internal class DataUploadRunnableTest {
             }
         }
     }
+
+    // region async
+
+    @Test
+    fun `𝕄 respect batch wait upload timeout 𝕎 run()`() {
+        // Given
+        whenever(mockStorage.readNextBatch(any(), any())) doAnswer {
+            // imitate async which never completes
+        }
+
+        // When
+        testedRunnable.run()
+
+        // Then
+        verify(mockThreadPoolExecutor).schedule(
+            same(testedRunnable),
+            any(),
+            eq(TimeUnit.MILLISECONDS)
+        )
+    }
+
+    @Test
+    fun `𝕄 stop waiting 𝕎 run() { exception is thrown }`(
+        @StringForgery batch: List<String>,
+        @StringForgery batchMeta: String,
+        forge: Forge
+    ) {
+        // Given
+        val batchId = mock<BatchId>()
+        val batchReader = mock<BatchReader>()
+        val batchData = batch.map { it.toByteArray() }
+        val batchMetadata = forge.aNullable { batchMeta.toByteArray() }
+
+        whenever(batchReader.read()) doReturn batchData
+        whenever(batchReader.currentMetadata()) doReturn batchMetadata
+
+        whenever(mockStorage.readNextBatch(any(), any())) doAnswer {
+            Thread {
+                it.getArgument<(BatchId, BatchReader) -> Unit>(1).invoke(batchId, batchReader)
+            }.start()
+        }
+
+        whenever(
+            mockDataUploader.upload(
+                fakeContext,
+                batchData,
+                batchMetadata
+            )
+        ) doThrow forge.aThrowable()
+
+        // When
+        val start = System.currentTimeMillis()
+        testedRunnable.run()
+
+        // Then
+        assertThat(System.currentTimeMillis() - start)
+            .isLessThan(TEST_BATCH_UPLOAD_WAIT_TIMEOUT_MS)
+        verify(mockThreadPoolExecutor).schedule(
+            same(testedRunnable),
+            any(),
+            eq(TimeUnit.MILLISECONDS)
+        )
+    }
+
+    // endregion
+
+    companion object {
+        const val TEST_BATCH_UPLOAD_WAIT_TIMEOUT_MS = 100L
+    }
 }