Add methods to clear our EvictingCircularBuffer (#1133)

* Add methods to clear our EvictingBuffer

* Fix test

* Bump patch version

* Changelog and version

* Fix test

* Make test less flaky

* Fix another flaky test

---------

Co-authored-by: Jason Leezer <[email protected]>

Author: jcleezer

CHANGELOG.md

@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.82.0] - 2026-01-13
+Add clear method to EvictingCircularBuffer that doesnt attempt to drain
+
 ## [29.81.2] - 2026-01-09
 - Clear DarkCluster Buffer on Refresh
 
@@ -5936,7 +5939,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.81.2...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.82.0...master
+[29.82.0]: https://github.com/linkedin/rest.li/compare/v29.81.2...v29.82.0
 [29.81.2]: https://github.com/linkedin/rest.li/compare/v29.81.1...v29.81.2
 [29.81.1]: https://github.com/linkedin/rest.li/compare/v29.81.0...v29.81.1
 [29.81.0]: https://github.com/linkedin/rest.li/compare/v29.80.3...v29.81.0

d2/src/test/java/com/linkedin/d2/balancer/servers/ZookeeperConnectionManagerTest.java

@@ -833,22 +833,54 @@ private static class IgnoreCancelledCallback implements Callback<None>
     @Override
     public void onError(Throwable e)
     {
-      if (e instanceof CancellationException || e.getCause() instanceof CancellationException || (e.getCause().getCause() != null && e.getCause()
-        .getCause() instanceof CancellationException))
-      {
+      if (isIgnorable(e)) {
         _callback.onSuccess(None.none());
+        return;
       }
-      else
-      {
-        _callback.onError(e);
-      }
+      _callback.onError(e);
     }
 
     @Override
     public void onSuccess(None result)
     {
       _callback.onSuccess(result);
     }
+
+    private boolean isIgnorable(Throwable t)
+    {
+      if (t == null)
+      {
+        return false;
+      }
+      // Directly ignorable types
+      if (t instanceof CancellationException)
+      {
+        return true;
+      }
+      if (t instanceof org.apache.zookeeper.KeeperException.NoNodeException)
+      {
+        return true;
+      }
+      // MultiException: all causes must be ignorable
+      if (t instanceof com.linkedin.common.callback.MultiException)
+      {
+        for (Throwable cause : ((com.linkedin.common.callback.MultiException) t).getCauses())
+        {
+          if (!isIgnorable(cause))
+          {
+            return false;
+          }
+        }
+        return true;
+      }
+      // Otherwise, walk the cause chain defensively
+      Throwable cause = t.getCause();
+      if (cause != null && cause != t)
+      {
+        return isIgnorable(cause);
+      }
+      return false;
+    }
   }
 
   private static void dataValidation(String uri, String cluster, double weight)

d2/src/test/java/com/linkedin/d2/discovery/event/PublisherTest.java

@@ -191,6 +191,9 @@ public void publishInitialize(String prop, String value)
       {
         _currentValues.put(prop, value);
         _initCondition.signalAll();
+        // Also signal add waiters: some publishers may emit initialize with a value
+        // (e.g., if a change occurs during initial fetch). Allow awaitAdd to wake and re-check.
+        _addCondition.signalAll();
       }
       finally
       {

darkcluster/src/main/java/com/linkedin/darkcluster/impl/ConstantQpsDarkClusterStrategy.java

@@ -84,7 +84,7 @@ public boolean handleRequest(RestRequest originalRequest, RestRequest darkReques
   @Override
   public void shutdown()
   {
-    _rateLimiter.cancelAll(new RuntimeException("Shutting down ConstantQpsDarkClusterStrategy"));
+    _rateLimiter.clear();
   }
 
   /**

darkcluster/src/test/java/com/linkedin/darkcluster/TestConstantQpsDarkClusterStrategy.java

@@ -204,11 +204,6 @@ public void testShutdown()
     // Call shutdown
     strategy.shutdown();
 
-    // Verify that cancelAll was called with a RuntimeException containing the expected message.
-    ArgumentCaptor<Throwable> throwableCaptor = ArgumentCaptor.forClass(Throwable.class);
-    verify(mockRateLimiter).cancelAll(throwableCaptor.capture());
-    Throwable throwable = throwableCaptor.getValue();
-    Assert.assertTrue(throwable instanceof RuntimeException);
-    Assert.assertEquals(throwable.getMessage(), "Shutting down ConstantQpsDarkClusterStrategy");
+    verify(mockRateLimiter).clear();
   }
 }

darkcluster/src/test/java/com/linkedin/darkcluster/TestDarkClusterStrategyFactory.java

@@ -323,16 +323,16 @@ public void testRefreshShutsDownExistingStrategiesPerPartition()
     // strategies were shut down.
     //
     // Note: RelativeTrafficMultiplierDarkClusterStrategy has no observable shutdown behavior. ConstantQpsDarkClusterStrategy
-    // does (it calls ConstantQpsRateLimiter.cancelAll), so we use CONSTANT_QPS here with a tracking rate limiter.
+    // does (it calls ConstantQpsRateLimiter.clear()), so we use CONSTANT_QPS here with a tracking rate limiter.
     MockClusterInfoProvider clusterInfoProvider = new MockClusterInfoProvider();
     Facilities facilities = new MockFacilities(clusterInfoProvider);
     DarkClusterDispatcher darkClusterDispatcher = new DefaultDarkClusterDispatcher(new MockClient(false));
     ClockedExecutor executor = new ClockedExecutor();
 
-    AtomicInteger cancelAllInvocations = new AtomicInteger(0);
+    AtomicInteger clearInvocations = new AtomicInteger(0);
     Supplier<ConstantQpsRateLimiter> rateLimiterSupplier = () ->
         new TrackingConstantQpsRateLimiter(executor, executor, executor,
-            TestConstantQpsDarkClusterStrategy.getBuffer(executor), cancelAllInvocations);
+            TestConstantQpsDarkClusterStrategy.getBuffer(executor), clearInvocations);
 
     DarkClusterStrategyFactory strategyFactory = new DarkClusterStrategyFactoryImpl(facilities,
         SOURCE_CLUSTER_NAME,
@@ -361,31 +361,32 @@ public void testRefreshShutsDownExistingStrategiesPerPartition()
 
     // Refresh again: should rebuild both partitions and shut down the old strategies.
     clusterInfoProvider.notifyListenersClusterAdded(SOURCE_CLUSTER_NAME);
-    Assert.assertEquals(cancelAllInvocations.get(), 2,
-        "Expected existing strategies to be shut down once per partition on refresh");
+    Assert.assertEquals(clearInvocations.get(), 2,
+        "Expected existing strategies to be cleared once per partition on refresh");
 
     Assert.assertTrue(strategyFactory.get(DARK_CLUSTER_NAME, 0) instanceof ConstantQpsDarkClusterStrategy);
     Assert.assertTrue(strategyFactory.get(DARK_CLUSTER_NAME, 1) instanceof ConstantQpsDarkClusterStrategy);
   }
 
   private static final class TrackingConstantQpsRateLimiter extends ConstantQpsRateLimiter
   {
-    private final AtomicInteger _cancelAllInvocations;
+    private final AtomicInteger _clearInvocations;
 
     TrackingConstantQpsRateLimiter(ScheduledExecutorService scheduler,
         java.util.concurrent.Executor executor,
         com.linkedin.util.clock.Clock clock,
         com.linkedin.r2.transport.http.client.EvictingCircularBuffer callbackBuffer,
-        AtomicInteger cancelAllInvocations)
+        AtomicInteger clearInvocations)
     {
       super(scheduler, executor, clock, callbackBuffer);
-      _cancelAllInvocations = cancelAllInvocations;
+      _clearInvocations = clearInvocations;
     }
 
     @Override
-    public void cancelAll(Throwable throwable)
+    public void clear()
     {
-      _cancelAllInvocations.incrementAndGet();
+      _clearInvocations.incrementAndGet();
+      super.clear();
     }
   }
 

gradle.properties

@@ -1,4 +1,4 @@
-version=29.81.2
+version=29.82.0
 group=com.linkedin.pegasus
 org.gradle.configureondemand=true
 org.gradle.parallel=true

r2-core/src/main/java/com/linkedin/r2/transport/http/client/ConstantQpsRateLimiter.java

@@ -65,6 +65,14 @@ public void setBufferTtl(int ttl, ChronoUnit ttlUnit)
     _evictingCircularBuffer.setTtl(ttl, ttlUnit);
   }
 
+  /**
+   * Clears all callbacks currently stored in the underlying {@link EvictingCircularBuffer}.
+   */
+  public void clear()
+  {
+    _evictingCircularBuffer.clear();
+  }
+
 
   private static class UnboundedRateLimiterExecutionTracker implements RateLimiterExecutionTracker
   {

r2-core/src/main/java/com/linkedin/r2/transport/http/client/EvictingCircularBuffer.java

@@ -156,6 +156,32 @@ int getCapacity()
     return _callbacks.size();
   }
 
+  /**
+   * Clears all callbacks and TTLs from the buffer immediately.
+   * Acquires write locks for all elements to avoid partially cleared reads/writes.
+   * Resets reader and writer positions to 0.
+   */
+  public void clear()
+  {
+    ArrayList<ReentrantReadWriteLock> tempLocks = new ArrayList<>();
+    _elementLocks.forEach(x ->
+    {
+      x.writeLock().lock();
+      tempLocks.add(x);
+    });
+    try
+    {
+      Collections.fill(_callbacks, null);
+      Collections.fill(_ttlBuffer, null);
+      _readerPosition.set(0);
+      _writerPosition.set(0);
+    }
+    finally
+    {
+      tempLocks.forEach(x -> x.writeLock().unlock());
+    }
+  }
+
   /**
    * Resizes the circular buffer, deleting the contents in the process.
    * This method should not be called frequently, ideally only as part of a startup lifecycle, as it does heavy locking

r2-core/src/test/java/com/linkedin/r2/transport/http/client/ratelimiter/TestConstantQpsRateLimiter.java

@@ -19,6 +19,7 @@
 import com.linkedin.common.callback.Callback;
 import com.linkedin.common.util.None;
 import com.linkedin.r2.transport.http.client.ConstantQpsRateLimiter;
+import com.linkedin.r2.transport.http.client.EvictingCircularBuffer;
 import com.linkedin.r2.transport.http.client.TestEvictingCircularBuffer;
 import com.linkedin.test.util.ClockedExecutor;
 import com.linkedin.test.util.retry.ThreeRetries;
@@ -28,9 +29,11 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.NoSuchElementException;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Assert;
 import org.testng.annotations.Test;
+import static org.junit.Assert.fail;
 
 
 public class TestConstantQpsRateLimiter
@@ -211,6 +214,59 @@ public void testLowRateHighlyParallelConsistentRandomness()
     assert(zeroFreqFailCount <= acceptableFailCount);
   }
 
+  @Test(timeOut = TEST_TIMEOUT)
+  /**
+   * This simulates the shutdown behavior of ConstantQpsDarkClusterStrategy,
+   * to verify that the buffer is cleared immediately upon shutdown.
+   */
+  public void shutdownClearsBufferImmediately()
+  {
+    ClockedExecutor executor = new ClockedExecutor();
+    // Use a separate clock for the circular buffer so TTL doesn’t auto-expire unless we advance it.
+    ClockedExecutor circularBufferExecutor = new ClockedExecutor();
+    EvictingCircularBuffer buffer = TestEvictingCircularBuffer.getBuffer(circularBufferExecutor);
+    ConstantQpsRateLimiter rateLimiter =
+        new ConstantQpsRateLimiter(executor, executor, executor, buffer);
+
+    // Ensure TTL is long so callbacks would normally remain unless explicitly cleared.
+    rateLimiter.setBufferTtl(Integer.MAX_VALUE, ChronoUnit.DAYS);
+    rateLimiter.setBufferCapacity(10);
+
+    // Submit several no-op callbacks
+    for (int i = 0; i < 10; i++)
+    {
+      rateLimiter.submit(new Callback<None>()
+      {
+        @Override
+        public void onError(Throwable e) { }
+
+        @Override
+        public void onSuccess(None result) { }
+      });
+    }
+
+    // Sanity: buffer should have something retrievable before cancelAll
+    Assert.assertNotNull(buffer.get());
+
+    // Invoke shutdown-like behavior: clear buffer immediately
+    rateLimiter.clear();
+
+    // Run the executor briefly to allow the event loop to process
+    executor.runFor(ONE_SECOND);
+
+    // Verify the buffer is cleared immediately (i.e., not waiting for TTL to expire)
+    boolean threw = false;
+    try
+    {
+      buffer.get();
+    }
+    catch (NoSuchElementException ex)
+    {
+      threw = true;
+    }
+    Assert.assertTrue("Expected the buffer to be empty immediately after clear()", threw);
+  }
+
   private static class TattlingCallback<T> implements Callback<T>
   {
     private AtomicInteger _interactCount = new AtomicInteger();