@Test
public void testFlush() throws Exception {

  final SettableApiFuture<Void> future1 = SettableApiFuture.create();
  final SettableApiFuture<Void> future2 = SettableApiFuture.create();
  when(batcher.add(rowMutation)).thenReturn(future1).thenReturn(future2);
  doAnswer(
          new Answer() {
            @Override
            public Object answer(InvocationOnMock invocationOnMock) {
              future1.set(null);
              future2.set(null);
              return null;
            }
          })
      .when(batcher)
      .flush();
  ApiFuture<Void> result1 = bulkMutationWrapper.add(rowMutation);
  ApiFuture<Void> result2 = bulkMutationWrapper.add(rowMutation);
  // flush should block until the responses are resolved.
  bulkMutationWrapper.flush();
  assertTrue(result1.isDone());
  assertTrue(result2.isDone());
  verify(batcher, times(2)).add(rowMutation);
  verify(batcher).flush();
}