@Test(timeout = CONSUME_TEST_TIMEOUT_MILLIS)
public void readAsyncTest() throws Exception {

    if (!consumer.isOpen()) {
        consumer.open();
    }
    ArrayList<String> ids = new ArrayList<>(200);
    for (int i = 0; i < 200; i++) {
        ids.add(Integer.toString(i));
    }
    ArrayList<CheramiFuture<CheramiDelivery>> futures = new ArrayList<>(200);
    for (int i = 0; i < 200; i++) {
        futures.add(consumer.readAsync());
    }
    for (int i = 0; i < 200; i++) {
        CheramiFuture<CheramiDelivery> future = futures.remove(0);
        CheramiDelivery delivery = future.get();
        String id = delivery.getMessage().getAckId();
        ids.remove(id);
    }
    assert (ids.isEmpty());
    //Test that the future is set to error.
    CheramiFuture<CheramiDelivery> future = consumer.readAsync();
    consumer.close();
    thrown.expect(ExecutionException.class);
    future.get();
    logger.info("CheramiConsumerImplTest: PASSED: ConsumeAsyncTest");
}