BufferedRateLimiter implements AsyncRateLimiter { @Override public void release() { permits.decrementAndGet(); totalReleased.incrementAndGet(); reprocessQueue(); } BufferedRateLimiter(@Value("${cassandra.query.buffer_size}") int queueLimit,
                               @Value("${cassandra.query.concurrent_limit}") int permitsLimit,
                               @Value("${cassandra.query.permit_max_wait_time}") int maxPermitWaitTime); @Override ListenableFuture<Void> acquireAsync(); @Override void release(); @Scheduled(fixedDelayString = "${cassandra.query.rate_limit_print_interval_ms}") void printStats();  }
@Test public void permitsReleasedInConcurrentMode() throws InterruptedException { BufferedRateLimiter limiter = new BufferedRateLimiter(10, 2, 100); AtomicInteger actualReleased = new AtomicInteger(); AtomicInteger actualRejected = new AtomicInteger(); ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5)); for (int i = 0; i < 100; i++) { ListenableFuture<ListenableFuture<Void>> submit = pool.submit(limiter::acquireAsync); Futures.addCallback(submit, new FutureCallback<ListenableFuture<Void>>() { @Override public void onSuccess(@Nullable ListenableFuture<Void> result) { Futures.addCallback(result, new FutureCallback<Void>() { @Override public void onSuccess(@Nullable Void result) { try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } limiter.release(); actualReleased.incrementAndGet(); } @Override public void onFailure(Throwable t) { actualRejected.incrementAndGet(); } }); } @Override public void onFailure(Throwable t) { } }); } TimeUnit.SECONDS.sleep(2); assertTrue("Unexpected released count " + actualReleased.get(), actualReleased.get() > 10 && actualReleased.get() < 20); assertTrue("Unexpected rejected count " + actualRejected.get(), actualRejected.get() > 80 && actualRejected.get() < 90); }