CheramiConsumerImpl implements CheramiConsumer, Reconfigurable { @Override public CheramiFuture<CheramiDelivery> readAsync() throws IOException { if (!isOpen()) { throw new IOException("Consumer closed."); } CheramiFuture<CheramiDelivery> future = new CheramiFuture<>(); futuresQueue.add(future); return future; } CheramiConsumerImpl(CheramiClient client, String dstPath, String consumerGroupName, ConsumerOptions options,
            MetricsReporter metricsReporter); @Override synchronized void open(); @Override synchronized void close(); @Override CheramiDelivery read(); @Override CheramiFuture<CheramiDelivery> readAsync(); @Override void ackDelivery(String deliveryToken); @Override void nackDelivery(String deliveryToken); boolean isOpen(); @Override void refreshNow();  }
@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()); CheramiFuture<CheramiDelivery> future = consumer.readAsync(); consumer.close(); thrown.expect(ExecutionException.class); future.get(); logger.info("CheramiConsumerImplTest: PASSED: ConsumeAsyncTest"); }