BatcherImpl implements Batcher<ElementT, ElementResultT> { @Override public void sendOutstanding() { final Batch<ElementT, ElementResultT, RequestT, ResponseT> accumulatedBatch; synchronized (elementLock) { if (currentOpenBatch.isEmpty()) { return; } accumulatedBatch = currentOpenBatch; currentOpenBatch = new Batch<>(prototype, batchingDescriptor, batchingSettings, batcherStats); } final ApiFuture<ResponseT> batchResponse = unaryCallable.futureCall(accumulatedBatch.builder.build()); numOfOutstandingBatches.incrementAndGet(); ApiFutures.addCallback( batchResponse, new ApiFutureCallback<ResponseT>() { @Override public void onSuccess(ResponseT response) { try { accumulatedBatch.onBatchSuccess(response); } finally { onBatchCompletion(); } } @Override public void onFailure(Throwable throwable) { try { accumulatedBatch.onBatchFailure(throwable); } finally { onBatchCompletion(); } } }, directExecutor()); } 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 testSendOutstanding() { final AtomicInteger callableCounter = new AtomicInteger(); underTest = new BatcherImpl<>( SQUARER_BATCHING_DESC_V2, new LabeledIntSquarerCallable() { @Override public ApiFuture<List<Integer>> futureCall(LabeledIntList request) { callableCounter.incrementAndGet(); return super.futureCall(request); } }, labeledIntList, batchingSettings, EXECUTOR); underTest.sendOutstanding(); assertThat(callableCounter.get()).isEqualTo(0); underTest.add(2); underTest.add(3); underTest.add(4); underTest.sendOutstanding(); assertThat(callableCounter.get()).isEqualTo(1); }