Removed SentryExecutorService limit (#4846)

* removed SentryExecutorService max queue limit

* updated changelog

* Format code

* format code

* readded executor queue limit for submit()
submit() now tries to purge if it reaches the limit

* updated changelog

* Format code

* added purge test

* Format code

* added purge test

---------

Co-authored-by: Sentry Github Bot <[email protected]>

Author: stefanosiano

CHANGELOG.md

@@ -18,6 +18,7 @@
 
 ### Fixes
 
+- Removed SentryExecutorService limit for delayed scheduled tasks ([#4846](https://github.com/getsentry/sentry-java/pull/4846))
 - [ANR] Removed AndroidTransactionProfiler lock ([#4817](https://github.com/getsentry/sentry-java/pull/4817))
 - Avoid ExecutorService for DefaultCompositePerformanceCollector timeout ([#4841](https://github.com/getsentry/sentry-java/pull/4841))
   - This avoids infinite data collection for never stopped transactions, leading to OOMs

sentry/src/main/java/io/sentry/SentryExecutorService.java

@@ -53,10 +53,19 @@ public SentryExecutorService() {
     this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), null);
   }
 
+  private boolean isQueueAvailable() {
+    // If limit is reached, purge cancelled tasks from the queue
+    if (executorService.getQueue().size() >= MAX_QUEUE_SIZE) {
+      executorService.purge();
+    }
+    // Check limit again after purge
+    return executorService.getQueue().size() < MAX_QUEUE_SIZE;
+  }
+
   @Override
   public @NotNull Future<?> submit(final @NotNull Runnable runnable)
       throws RejectedExecutionException {
-    if (executorService.getQueue().size() < MAX_QUEUE_SIZE) {
+    if (isQueueAvailable()) {
       return executorService.submit(runnable);
     }
     if (options != null) {
@@ -70,7 +79,7 @@ public SentryExecutorService() {
   @Override
   public @NotNull <T> Future<T> submit(final @NotNull Callable<T> callable)
       throws RejectedExecutionException {
-    if (executorService.getQueue().size() < MAX_QUEUE_SIZE) {
+    if (isQueueAvailable()) {
       return executorService.submit(callable);
     }
     if (options != null) {
@@ -84,15 +93,7 @@ public SentryExecutorService() {
   @Override
   public @NotNull Future<?> schedule(final @NotNull Runnable runnable, final long delayMillis)
       throws RejectedExecutionException {
-    if (executorService.getQueue().size() < MAX_QUEUE_SIZE) {
-      return executorService.schedule(runnable, delayMillis, TimeUnit.MILLISECONDS);
-    }
-    if (options != null) {
-      options
-          .getLogger()
-          .log(SentryLevel.WARNING, "Task " + runnable + " rejected from " + executorService);
-    }
-    return new CancelledFuture<>();
+    return executorService.schedule(runnable, delayMillis, TimeUnit.MILLISECONDS);
   }
 
   @Override

sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt

@@ -6,6 +6,7 @@ import java.util.concurrent.Callable
 import java.util.concurrent.CancellationException
 import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.ScheduledThreadPoolExecutor
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -177,27 +178,6 @@ class SentryExecutorServiceTest {
     verify(executor).submit(any<Callable<String>>())
   }
 
-  @Test
-  fun `SentryExecutorService schedule returns cancelled future when queue size exceeds limit`() {
-    val queue = mock<BlockingQueue<Runnable>>()
-    whenever(queue.size).thenReturn(272) // Above MAX_QUEUE_SIZE (271)
-
-    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
-
-    val options = mock<SentryOptions>()
-    val logger = mock<ILogger>()
-    whenever(options.logger).thenReturn(logger)
-
-    val sentryExecutor = SentryExecutorService(executor, options)
-    val future = sentryExecutor.schedule({}, 1000L)
-
-    assertTrue(future.isCancelled)
-    assertTrue(future.isDone)
-    assertFailsWith<CancellationException> { future.get() }
-    verify(executor, never()).schedule(any<Runnable>(), any(), any())
-    verify(logger).log(any<SentryLevel>(), any<String>())
-  }
-
   @Test
   fun `SentryExecutorService schedule accepts when queue size is within limit`() {
     val queue = mock<BlockingQueue<Runnable>>()
@@ -225,4 +205,28 @@ class SentryExecutorServiceTest {
     // the queue should be empty
     assertEquals(0, executor.queue.size)
   }
+
+  @Test
+  fun `SentryExecutorService schedules any number of job`() {
+    val executor = ScheduledThreadPoolExecutor(1)
+    val sentryExecutor = SentryExecutorService(executor, null)
+    // Post 1k jobs after 1 day, to test they are all accepted
+    repeat(1000) { sentryExecutor.schedule({}, TimeUnit.DAYS.toMillis(1)) }
+    assertEquals(1000, executor.queue.size)
+  }
+
+  @Test
+  fun `SentryExecutorService purges cancelled jobs when limit is reached`() {
+    val executor = ScheduledThreadPoolExecutor(1)
+    val sentryExecutor = SentryExecutorService(executor, null)
+    // Post 1k jobs after 1 day, to test they are all accepted
+    repeat(1000) {
+      val future = sentryExecutor.schedule({}, TimeUnit.DAYS.toMillis(1))
+      future.cancel(true)
+    }
+    assertEquals(1000, executor.queue.size)
+    // Submit should purge cancelled scheduled jobs
+    sentryExecutor.submit {}
+    assertEquals(1, executor.queue.size)
+  }
 }