perf: Schedule rate-limit notifications on shared executor (JAVA-653)

RateLimiter created a java.util.Timer whose thread stayed alive forever
once the SDK got rate limited. Schedule the "rate limit lifted"
observer notification on the shared timer executor instead, whose
single worker thread is reused across all timeouts and self-terminates
when idle. Pending notifications are cancelled on close().

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: runningcode

sentry/src/main/java/io/sentry/transport/RateLimiter.java

@@ -23,12 +23,13 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -42,8 +43,9 @@ public final class RateLimiter implements Closeable {
   private final @NotNull Map<DataCategory, @NotNull Date> sentryRetryAfterLimit =
       new ConcurrentHashMap<>();
   private final @NotNull List<IRateLimitObserver> rateLimitObservers = new CopyOnWriteArrayList<>();
-  private @Nullable Timer timer = null;
-  private final @NotNull AutoClosableReentrantLock timerLock = new AutoClosableReentrantLock();
+  private final @NotNull List<Future<?>> notifyObserversFutures = new ArrayList<>();
+  private final @NotNull AutoClosableReentrantLock notifyFuturesLock =
+      new AutoClosableReentrantLock();
 
   public RateLimiter(
       final @NotNull ICurrentDateProvider currentDateProvider,
@@ -278,11 +280,11 @@ public void updateRetryAfterLimits(
                   continue;
                 }
 
-                applyRetryAfterOnlyIfLonger(dataCategory, date);
+                applyRetryAfterOnlyIfLonger(dataCategory, date, retryAfterMillis);
               }
             } else {
               // if categories are empty, we should apply to "all" categories.
-              applyRetryAfterOnlyIfLonger(DataCategory.All, date);
+              applyRetryAfterOnlyIfLonger(DataCategory.All, date, retryAfterMillis);
             }
           }
         }
@@ -291,7 +293,7 @@ public void updateRetryAfterLimits(
       final long retryAfterMillis = parseRetryAfterOrDefault(retryAfterHeader);
       // we dont care if Date is UTC as we just add the relative seconds
       final Date date = new Date(currentDateProvider.getCurrentTimeMillis() + retryAfterMillis);
-      applyRetryAfterOnlyIfLonger(DataCategory.All, date);
+      applyRetryAfterOnlyIfLonger(DataCategory.All, date, retryAfterMillis);
     }
   }
 
@@ -300,10 +302,11 @@ public void updateRetryAfterLimits(
    *
    * @param dataCategory the DataCategory
    * @param date the Date to be applied
+   * @param delayMillis the millis until the rate limit is lifted
    */
   @SuppressWarnings({"JdkObsolete", "JavaUtilDate"})
   private void applyRetryAfterOnlyIfLonger(
-      final @NotNull DataCategory dataCategory, final @NotNull Date date) {
+      final @NotNull DataCategory dataCategory, final @NotNull Date date, final long delayMillis) {
     final Date oldDate = sentryRetryAfterLimit.get(dataCategory);
 
     // only overwrite its previous date if the limit is even longer
@@ -312,19 +315,25 @@ private void applyRetryAfterOnlyIfLonger(
 
       notifyRateLimitObservers();
 
-      try (final @NotNull ISentryLifecycleToken ignored = timerLock.acquire()) {
-        if (timer == null) {
-          timer = new Timer(true);
+      // notify observers again once the rate limit is lifted, using the shared timer executor
+      // instead of a dedicated Timer thread
+      try (final @NotNull ISentryLifecycleToken ignored = notifyFuturesLock.acquire()) {
+        final @NotNull Iterator<Future<?>> iterator = notifyObserversFutures.iterator();
+        while (iterator.hasNext()) {
+          if (iterator.next().isDone()) {
+            iterator.remove();
+          }
+        }
+        try {
+          notifyObserversFutures.add(
+              options
+                  .getTimerExecutorService()
+                  .schedule(() -> notifyRateLimitObservers(), delayMillis));
+        } catch (RejectedExecutionException e) {
+          options
+              .getLogger()
+              .log(SentryLevel.WARNING, "Failed to schedule rate limit lifted notification.", e);
         }
-
-        timer.schedule(
-            new TimerTask() {
-              @Override
-              public void run() {
-                notifyRateLimitObservers();
-              }
-            },
-            date);
       }
     }
   }
@@ -364,11 +373,11 @@ public void removeRateLimitObserver(@NotNull final IRateLimitObserver observer)
 
   @Override
   public void close() throws IOException {
-    try (final @NotNull ISentryLifecycleToken ignored = timerLock.acquire()) {
-      if (timer != null) {
-        timer.cancel();
-        timer = null;
+    try (final @NotNull ISentryLifecycleToken ignored = notifyFuturesLock.acquire()) {
+      for (Future<?> future : notifyObserversFutures) {
+        future.cancel(false);
       }
+      notifyObserversFutures.clear();
     }
     rateLimitObservers.clear();
   }

sentry/src/test/java/io/sentry/transport/RateLimiterTest.kt

@@ -18,6 +18,7 @@ import io.sentry.SentryEnvelope
 import io.sentry.SentryEnvelopeHeader
 import io.sentry.SentryEnvelopeItem
 import io.sentry.SentryEvent
+import io.sentry.SentryExecutorService
 import io.sentry.SentryLogEvent
 import io.sentry.SentryLogEvents
 import io.sentry.SentryLogLevel
@@ -37,11 +38,10 @@ import io.sentry.protocol.SentryId
 import io.sentry.protocol.SentryTransaction
 import io.sentry.protocol.User
 import io.sentry.test.getProperty
-import io.sentry.test.injectForField
 import io.sentry.util.HintUtils
 import java.io.File
-import java.util.Timer
 import java.util.UUID
+import java.util.concurrent.Future
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -66,6 +66,8 @@ class RateLimiterTest {
 
     fun getSUT(): RateLimiter {
       val options = SentryOptions().apply { setLogger(NoOpLogger.getInstance()) }
+      // a real executor so scheduled rate-limit-lifted notifications actually run
+      options.setTimerExecutorService(SentryExecutorService(options))
 
       SentryOptionsManipulator.setClientReportRecorder(options, clientReportRecorder)
 
@@ -654,7 +656,7 @@ class RateLimiterTest {
   }
 
   @Test
-  fun `apply rate limits schedules a timer to notify observers of lifted limits`() {
+  fun `apply rate limits schedules a task to notify observers of lifted limits`() {
     val rateLimiter = fixture.getSUT()
     whenever(fixture.currentDateProvider.currentTimeMillis).thenReturn(0, 1, 2001)
 
@@ -667,18 +669,19 @@ class RateLimiterTest {
   }
 
   @Test
-  fun `close cancels the timer`() {
+  fun `close cancels pending notify tasks`() {
     val rateLimiter = fixture.getSUT()
-    val timer = mock<Timer>()
-    rateLimiter.injectForField("timer", timer)
+    rateLimiter.updateRetryAfterLimits("60:replay:key", null, 1)
+
+    val futures = rateLimiter.getProperty<List<Future<*>>>("notifyObserversFutures")
+    assertEquals(1, futures.size)
+    val future = futures.first()
 
     // When the rate limiter is closed
     rateLimiter.close()
 
-    // Then the timer is cancelled
-    verify(timer).cancel()
-
-    // And is removed by the rateLimiter
-    assertNull(rateLimiter.getProperty("timer"))
+    // Then the pending notify task is cancelled and dropped
+    assertTrue(future.isCancelled)
+    assertTrue(rateLimiter.getProperty<List<Future<*>>>("notifyObserversFutures").isEmpty())
   }
 }