code: protected void createConsumer(final Map<String, Object> config) {
          final Properties consumerConfig = createKafkaConfig(config);
  
          LOG.info("connecting kafka client to zookeeper at {} as client group {}",
              consumerConfig.getProperty("zookeeper.connect"),
              consumerConfig.getProperty("group.id"));
          _consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerConfig));
      }
docs: Ensures an initialized kafka {@link ConsumerConnector} is present.
  
  @param config The storm configuration passed to {@link #open(Map, TopologyContext, SpoutOutputCollector)}.
  @throws IllegalArgumentException When a required configuration parameter is missing or a sanity check fails.