@Override
    public void subscribe(Collection<String> topics) {
        subscribeInternal(topics, Optional.empty());
    }
@Test
    public void testSubscriptionOnEmptyTopic() {
        consumer = newConsumer();
        String emptyTopic = "  ";
        assertThrows(IllegalArgumentException.class, () -> consumer.subscribe(singletonList(emptyTopic)));
    }