title: Storm Bolt not printing/logging Kafka Spout
body: <p>Edit: I added an .ack() to the Bolt (which required me to use a Rich Bolt instead of the basic bolt) and am having the same issue - nothing that tells me tuples are being processed by the bolt.</p>
  
  <p>If it matters, I'm running this on a CentOS image on an EC2 instance. Any help would be appreciated.</p>
  
  <hr>
  
  <hr>
  
  <p>I'm trying to set up a very basic HelloWorld Storm example to read messages from a Kafka cluster and print/log the messages I get.</p>
  
  <p>Currently I have 20 messages in the Kafka cluster. When I run the topology (which appears to start just fine), I am able to see my Kafka Spout as well as the Echo Bolt.  In the Storm UI, the Kafka Spout <code>Acked</code> column has 20 as a value - which I would assume is the number of messages that it was able to read/access (?)</p>
  
  <p>The Echo Bolt line, however, only notes that I have 1 executor and 1 tasks. All other columns are 0.</p>
  
  <p>Looking at the Storm worker log that is generated, I see this line: <code>Read partition information from: /HelloWorld Spout/partition_0  --&gt; {"topic":"helloworld","partition":0,"topology":{"id":"&lt;UUID&gt;","name":"Kafka-Storm test"},"broker":{"port":6667,"host":"ip-10-0-0-35.ec2.internal"},"offset":20}</code></p>
  
  <p>The next few lines are as follows:</p>
  
  <pre><code>s.k.PartitionManager [INFO] Last commit offset from zookeeper: 0
  s.k.PartitionManager [INFO] Commit offset 0 is more than 9223372036854775807 behind, resetting to startOffsetTime=-2
  s.k.PartitionManager [INFO] Starting Kafka ip-10-0-0-35.ec2.internal:0 from offset 0
  s.k.ZkCoordinator [INFO] Task [1/1] Finished refreshing
  s.k.ZkCoordinator [INFO] Task [1/1] Refreshing partition manager connections
  s.k.DynamicBrokersReader [INFO] Read partition info from zookeeper: GlobalPartitionInformation{partitionMap={0=ip-10-0-0-35.ec2.internal:6667}}
  </code></pre>
  
  <p>The rest of the worker log shows no log/print out of the messages processed by the Bolt. I'm at a loss of why the Bolt doesn't seem to be getting any of the messages from the Kafka Cluster. Any help would be great. Thanks.</p>
  
  <p><strong>Building the KafkaSpout</strong></p>
  
  <pre><code>private static KafkaSpout setupSpout() {
    BrokerHosts hosts = new ZkHosts("localhost:2181");
    SpoutConfig spoutConfig = new SpoutConfig(hosts, "helloworld", "", "HelloWorld Spout");
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    spoutConfig.forceFromStart = true;
    spoutConfig.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
    return new KafkaSpout(spoutConfig);
  }
  </code></pre>
  
  <p><strong>Building the topology and submitting it</strong></p>
  
  <pre><code>public static void main(String[] args) {
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("Kafka Spout", setupSpout());
    builder.setBolt("Echo Bolt", new SystemOutEchoBolt());
  
    try {
      System.setProperty("storm.jar", "/tmp/storm.jar");
      StormSubmitter.submitTopology("Kafka-Storm test", new Config(), builder.createTopology());
    } //catchExceptionsHere
  }
  </code></pre>
  
  <p><strong>Bolt</strong></p>
  
  <pre><code>public class SystemOutEchoBolt extends BaseRichBolt {
  
    private static final long serialVersionUID = 1L;
    private static final Logger logger = LoggerFactory.getLogger(SystemOutEchoBolt.class);
  
    private OutputCollector m_collector;
  
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map _map, TopologyContext _conetxt, OutputCollector _collector) {
      m_collector = _collector;
    }
  
    @Override
    public void execute(Tuple _tuple) {
      System.out.println("Printing tuple with toString(): " + _tuple.toString());
      System.out.println("Printing tuple with getString(): " + _tuple.getString(0));
      logger.info("Logging tuple with logger: " + _tuple.getString(0));
      m_collector.ack(_tuple);
    }
  
    @Override
    public void declareOutputFields(OutputFieldsDeclarer _declarer) {}
  }
  </code></pre>