code: protected KafkaClient buildKafkaClient() throws Exception {
          if (StringUtils.isBlank(bootstrapServers)) {
              throw new IllegalStateException("Kafka bootstrap server list is not defined.");
          }
          KafkaClient kafkaClient = new KafkaClient(bootstrapServers);
          kafkaClient.setProducerProperties(consumerProps).setConsumerProperties(consumerProps);
          kafkaClient.init();
          return kafkaClient;
      }
docs: /*----------------------------------------------------------------------