@Override
    public ConsumerBuilder<T> topics(List<String> topicNames) {
        checkArgument(topicNames != null && !topicNames.isEmpty(),
                "Passed in topicNames list should not be null or empty.");
        topicNames.stream().forEach(topicName ->
                checkArgument(StringUtils.isNotBlank(topicName), "topicNames cannot have blank topic"));
        conf.getTopicNames().addAll(topicNames.stream().map(StringUtils::trim).collect(Collectors.toList()));
        return this;
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void testConsumerBuilderImplWhenTopicNamesHasBlankTopic() {
        List<String> topicNames = Arrays.asList("my-topic", " ");
        consumerBuilderImpl.topics(topicNames);
    }