CheramiConsumerImpl implements CheramiConsumer, Reconfigurable { @Override public CheramiDelivery read() throws InterruptedException, IOException { if (!isOpen()) { throw new IOException("Consumer closed."); } CheramiFuture<CheramiDelivery> future = readAsync(); try { return future.get(); } catch (ExecutionException e) { throw new IOException(e); } } 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 readTest() 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)); } for (int i = 0; i < 200; i++) { CheramiDelivery delivery = consumer.read(); String id = delivery.getMessage().getAckId(); ids.remove(id); } assert (ids.isEmpty()); thrown.expect(TestTimedOutException.class); consumer.read(); }