@Test
public void test() throws Exception {
    ConcurrentMessageListenerContainer<?, ?> container = (ConcurrentMessageListenerContainer<?, ?>) registry
            .getListenerContainer("EmptyLegCommandFeed_V2");
    container.stop();
    @SuppressWarnings("unchecked")
    AcknowledgingConsumerAwareMessageListener<String, String> messageListener = (AcknowledgingConsumerAwareMessageListener<String, String>) container
            .getContainerProperties().getMessageListener();
    CountDownLatch latch = new CountDownLatch(1);
    container.getContainerProperties()
            .setMessageListener(new AcknowledgingConsumerAwareMessageListener<String, String>() {

                @Override
                public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment,
                                      Consumer<?, ?> consumer) {
                    messageListener.onMessage(data, acknowledgment, consumer);
                    latch.countDown();
                }

            });
    container.start();
    template.send("EmptyLegCommandFeed_V2", "foo");
    assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
}