### Question:
ThrottledAsyncQueue extends AsyncQueue<T> { @Override public synchronized <O> ListenableFuture<O> borrowBatchAsync(int maxSize, Function<List<T>, BorrowResult<T, O>> function) { checkArgument(maxSize >= 0, "maxSize must be at least 0"); ListenableFuture<?> throttleFuture = immediateFuture(null); if (size() > 0) { int size = maxBatchSize(maxSize); if (rateLimiter.tryAcquire(size)) { return super.borrowBatchAsync(size, function); } } else if (!isFinished()) { throttleFuture = getNotEmptySignal(); } return Futures.transformAsync( throttleFuture, any -> { int size = maxBatchSize(maxSize); if (size > 0) { rateLimiter.acquire(size); } return super.borrowBatchAsync(size, function); }, executor); } ThrottledAsyncQueue(int maxBatchSizePerSec, int targetQueueSize, Executor executor); @Override synchronized ListenableFuture<O> borrowBatchAsync(int maxSize, Function<List<T>, BorrowResult<T, O>> function);  }### Answer:
@Test(timeOut = 10_000) public void testBorrowThrows() throws Exception { AsyncQueue<Integer> queue = new ThrottledAsyncQueue<>(100, 4, executor); queue.offer(1); queue.offer(2); queue.offer(3); queue.offer(4); queue.offer(5); ListenableFuture<?> future1 = queue.offer(6); assertFalse(future1.isDone()); Runnable runnable = () -> { getFutureValue(queue.borrowBatchAsync(1, elements -> { throw new RuntimeException("test fail"); })); }; assertThatThrownBy(() -> executor.submit(runnable).get()) .isInstanceOf(ExecutionException.class) .hasMessageContaining("test fail"); ListenableFuture<?> future2 = queue.offer(7); assertFalse(future1.isDone()); assertFalse(future2.isDone()); queue.finish(); future1.get(); future2.get(); assertTrue(queue.offer(8).isDone()); assertThatThrownBy(() -> executor.submit(runnable).get()) .isInstanceOf(ExecutionException.class) .hasMessageContaining("test fail"); assertTrue(queue.offer(9).isDone()); assertFalse(queue.isFinished()); assertEquals(queue.getBatchAsync(100).get(), ImmutableList.of(3, 4, 5, 6, 7)); assertTrue(queue.isFinished()); }