CheramiConsumerImpl implements CheramiConsumer, Reconfigurable { @Override public synchronized void open() throws InterruptedException, IOException { if (!isOpen()) { String clientName = this.client.getOptions().getClientAppName(); this.tChannel = new TChannel.Builder(String.format("%s_%s", clientName, consumerGroupName)).build(); this.subChannel = tChannel.makeSubChannel(OUTPUT_SERVICE_NAME); connectionManager.open(); new Thread(deliveryRunnable).start(); isOpen.set(true); logger.info("Consumer opened."); } } 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 public void openTest() throws Exception { consumer.open(); assert (consumer.isOpen() == true); consumer.close(); logger.info("CheramiConsumerImplTest: PASSED: OpenTest"); }