@Test(timeout = 1000)
public void batcher_batches_upto_maxBatchSize_under_pressure() throws Exception {

  List<List<Integer>> batches = new ArrayList<>();
  Function<List<Integer>, List<Integer>> batchHandler = l -> {
    batches.add(l);
    return l.stream().map(i -> i * 10).collect(toList());
  };
  AtomicReference<Runnable> processQueue = new AtomicReference<>();
  ThreadFactory factory = r -> {
    processQueue.set(r);
    return new Thread();
  };
  Thread t = null;
  try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(5, 2, EnqueueBehaviour.COMPLETE_EXCEPTIONALLY,
      batchHandler, factory)) {
    CompletableFuture<Integer> result1 = batcher.apply(1);
    CompletableFuture<Integer> result2 = batcher.apply(2);
    CompletableFuture<Integer> result3 = batcher.apply(3);
    CompletableFuture<Integer> result4 = batcher.apply(4);
    CompletableFuture<Integer> result5 = batcher.apply(5);
    t = new Thread(processQueue.get());
    t.start();
    assertThat(result1.get(), is(10));
    assertThat(result2.get(), is(20));
    assertThat(result3.get(), is(30));
    assertThat(result4.get(), is(40));
    assertThat(result5.get(), is(50));
  }
  t.join();
  assertThat(batches.size(), is(3));
  assertThat(batches.get(0).size(), is(2));
  assertThat(batches.get(0), contains(1, 2));
  assertThat(batches.get(1).size(), is(2));
  assertThat(batches.get(1), contains(3, 4));
  assertThat(batches.get(2).size(), is(1));
  assertThat(batches.get(2), contains(5));
}