title: Spout not getting acked
body: <p>In one of our topologies containing 1 spout and 1 bolt - I had a hunch that bolts are completing fine (and acking) but spouts are still failing.</p>
  <p>I tried to confirm this by having a TaskHook as below -</p>
  <pre><code>public class BaseHook extends BaseTaskHook {
  
      private Logger logger;
      private String topology;
      private String component;
  
      public BaseHook(String component) {
          this.component = component;
      }
  
      @Override
      public void prepare(Map conf, TopologyContext context) {
          logger = LoggerFactory.getLogger(this.getClass());
          this.topology = (String) conf.get(&quot;topology.name&quot;);
      }
  
      @Override
      public void emit(EmitInfo info) {
          log(&quot;EMITTED &gt;&gt; Value = &quot; + info.values);
      }
  
      @Override
      public void spoutAck(SpoutAckInfo info) {
          log(&quot;ACKED &gt;&gt; Tuple = &quot; + info.messageId + &quot;, Latency = &quot; + info.completeLatencyMs);
      }
  
      @Override
      public void spoutFail(SpoutFailInfo info) {
          log(&quot;FAILED &gt;&gt; Tuple = &quot; + info.messageId + &quot;, Latency = &quot; + info.failLatencyMs);
      }
  
      @Override
      public void boltExecute(BoltExecuteInfo info) {
          log(&quot;EXECUTED &gt;&gt; Tuple = &quot; + info.tuple.getValues() + &quot;, Latency = &quot; + info.executeLatencyMs);
      }
  
      @Override
      public void boltAck(BoltAckInfo info) {
          log(&quot;ACKED &gt;&gt; Tuple = &quot; + info.tuple.getValues() + &quot;, Latency = &quot; + info.processLatencyMs);
      }
  
      @Override
      public void boltFail(BoltFailInfo info) {
          log(&quot;FAILED &gt;&gt; Tuple = &quot; + info.tuple.getValues() + &quot;, Latency = &quot; + info.failLatencyMs);
      }
  
      private void log(String msg) {
          logger.info(&quot;&gt;&gt;&gt;&gt;&gt; &quot; + topology + &quot; &gt;&gt; &quot; + component + &quot; &gt;&gt; &quot; + msg);
      }
  }
  </code></pre>
  <p>Turned out that my hunch is correct. The logs look something like this -</p>
  <pre><code>&gt;&gt;&gt;&gt;&gt; TopologyX &gt;&gt; SpoutX &gt;&gt; EMITTED &gt;&gt; Value = [XXXXXXXXX]
  &gt;&gt;&gt;&gt;&gt; TopologyX &gt;&gt; BoltX &gt;&gt; ACKED &gt;&gt; Tuple = [XXXXXXXXX], Latency = 1972
  &gt;&gt;&gt;&gt;&gt; TopologyX &gt;&gt; BoltX &gt;&gt; EXECUTED &gt;&gt; Tuple = [XXXXXXXXX], Latency = 1973
  &gt;&gt;&gt;&gt;&gt; TopologyX &gt;&gt; SpoutX &gt;&gt; FAILED &gt;&gt; Tuple = XXXXXXXXX, Latency = 53913
  </code></pre>
  <p>ie. The Bolt is taking almost 2s (To Execute and Ack) but Spout Fail is being called at around 53s (almost twice of <code>topology.message.timeout.secs * 2</code>.</p>
  <p>I would expect Spout Ack to be called that too within 2-3sec. Spout is non-blocking and bolts and ackers both have ample capacity to work with.</p>
  <p>Anyone got any hints as to what might be the cause?</p>
  <hr />
  <h2>Update</h2>
  <p>So here is what out storm cluster looked like -</p>
  <ul>
  <li>4 Topologies
  <ul>
  <li><code>T1</code> = S &gt; B &gt; B &gt; B &gt; Ack/Fail</li>
  <li><code>T2</code> = S &gt; B &gt; Ack/Fail</li>
  <li><code>T3</code> = S &gt; B &gt; B &gt; Ack/Fail</li>
  <li><code>T4</code> =
  <ul>
  <li>S &gt; B &gt; Ack/Fail</li>
  <li>S &gt; B &gt; Ack/Fail</li>
  </ul>
  </li>
  </ul>
  </li>
  </ul>
  <p>So, the topology in question is <code>T4</code> ie. the one with 2 different spouts and 2 bolts. One of these flows, generally works fine (they do have different messageIds uniquely identifying tuples)</p>
  <p>Can this be the issue?</p>
  <p>Anyways,</p>
  <ul>
  <li>We tried reducing executors to whatever we can but that didn't improve anything in <code>T4</code>.</li>
  <li>We disabled all other topologies and things worked absolutely fine for <code>T4</code></li>
  <li>We enabled <code>T1</code> and still things worked fine</li>
  <li>We enabled <code>T2</code> (and <code>T3</code> on other occasion) and <code>T4</code> started failing</li>
  </ul>
  <p>Now,</p>
  <ul>
  <li>On one random occasion, <code>T4</code> even worked with T1 and T3 both enabled.</li>
  <li>But otherwise, every time <code>T2</code> or <code>T3</code> is enabled, <code>T4</code> breaks down.</li>
  </ul>
  <p>Points to note -</p>
  <ul>
  <li><code>T3</code> and <code>T4</code> are both quick topologies ie. their flow finishes in &lt; 100ms</li>
  <li>Both, <code>T3</code> and <code>T4</code> have just 1 executor per Spout and Bolts</li>
  <li>Both, <code>T3</code> and <code>T4</code> have Max Tuple Pending = 1</li>
  <li>We want to rate limit both <code>T3</code> and <code>T4</code> (but have tried without rate limiting too)
  <ul>
  <li>Try 1: Without any limiting</li>
  <li>Try 2: Sleep for 50ms before emitting</li>
  <li>Try 3: Sleep for 50ms after emitting</li>
  <li>Try 4: Don't sleep, but emit only if its been 50s from last emit</li>
  <li>Nothing worked</li>
  </ul>
  </li>
  </ul>
  <hr />
  <h2>Additional Info based on comment</h2>
  <p>All out Spouts extend from BaseSpout class -</p>
  <pre><code>public abstract class BaseSpout extends BaseRichSpout {
      private SpoutOutputCollector collector;
  
      @Override
      public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
          context.addTaskHook(new BaseHook(this.getClass().getSimpleName()));
          try {
              this.collector = collector;
              open();
          } catch (Exception e) {
              throw new RuntimeException(&quot;Error when preparing spout&quot;, e);
          }
      }
  
      @Override
      public void nextTuple() {
          try {
              getTuple();
          } catch (Throwable t) {
              if (!(t instanceof FailedException)) {
                  t = new FailedException(&quot;nextTuple()&quot;, t);
              }
              collector.reportError(t);
          }
      }
  
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          String[] fields = getFields();
          if (fields != null) {
              declarer.declare(new Fields(fields));
          }
      }
  
      protected void emit(Values values, String msgId) {
          collector.emit(values, msgId);
      }
  
      protected abstract void open() throws Exception;
  
      protected abstract void getTuple() throws Exception;
  
      protected abstract String[] getFields();
  }
  </code></pre>
  <p>and all out bolts extend from BaseBolt class -</p>
  <pre><code>public abstract class BaseBolt extends BaseRichBolt {
  
      private OutputCollector collector;
  
      @Override
      public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {   
          context.addTaskHook(new BaseHook(this.getClass().getSimpleName()));
          try {
              this.collector = collector;
              prepare();
          } catch (Exception e) {
              throw new RuntimeException(&quot;Error when preparing bolt&quot;, e);
          }
      }
  
      @Override
      public void execute(Tuple tuple) {
          try {
              process(tuple);
              collector.ack(tuple);
          } catch (Throwable t) {
              if (!(t instanceof FailedException)) {
                  t = new FailedException(&quot;execute(&quot; + tuple + &quot;)&quot;, t);
              }
              collector.reportError(t);
              collector.fail(tuple);
          }
      }
  
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          String[] fields = getFields();
          if (fields != null) {
              declarer.declare(new Fields(fields));
          }
      }
  
      protected void emit(Tuple tuple, Values values) {
          collector.emit(tuple, values);
      }
  
      protected abstract void prepare() throws Exception;
  
      protected abstract void process(Tuple tuple) throws Exception;
  
      protected abstract String[] getFields();
  }
  </code></pre>
  <p>So to say, there is no possibility of emitting a tuple without messageID (from spout) or unanchored tuple (from bolt)</p>