BatcherImpl implements Batcher<ElementT, ElementResultT> { @Override public void close() throws InterruptedException { if (isClosed) { return; } flush(); scheduledFuture.cancel(true); isClosed = true; currentBatcherReference.closed = true; currentBatcherReference.clear(); BatchingException exception = batcherStats.asException(); if (exception != null) { throw exception; } } BatcherImpl(
      BatchingDescriptor<ElementT, ElementResultT, RequestT, ResponseT> batchingDescriptor,
      UnaryCallable<RequestT, ResponseT> unaryCallable,
      RequestT prototype,
      BatchingSettings batchingSettings,
      ScheduledExecutorService executor); @Override ApiFuture<ElementResultT> add(ElementT element); @Override void flush(); @Override void sendOutstanding(); @Override void close();  }
@Test public void testPushCurrentBatchRunnable() throws Exception { long DELAY_TIME = 50L; BatchingSettings settings = batchingSettings.toBuilder().setDelayThreshold(Duration.ofMillis(DELAY_TIME)).build(); BatcherImpl<Integer, Integer, LabeledIntList, List<Integer>> batcher = new BatcherImpl<>( SQUARER_BATCHING_DESC_V2, callLabeledIntSquarer, labeledIntList, settings, EXECUTOR); BatcherImpl.PushCurrentBatchRunnable<Integer, Integer, LabeledIntList, List<Integer>> pushBatchRunnable = new BatcherImpl.PushCurrentBatchRunnable<>(batcher); ScheduledFuture<?> onGoingRunnable = EXECUTOR.scheduleWithFixedDelay( pushBatchRunnable, DELAY_TIME, DELAY_TIME, TimeUnit.MILLISECONDS); pushBatchRunnable.setScheduledFuture(onGoingRunnable); boolean isExecutorCancelled = pushBatchRunnable.isCancelled(); assertThat(isExecutorCancelled).isFalse(); batcher.close(); batcher = null; for (int retry = 0; retry < 3; retry++) { System.gc(); System.runFinalization(); isExecutorCancelled = pushBatchRunnable.isCancelled(); if (isExecutorCancelled) { break; } Thread.sleep(DELAY_TIME * (1L << retry)); } assertThat(pushBatchRunnable.isCancelled()).isTrue(); }