title: Invalid Topology Exception error while submitting a topology
body: <p>While working on my first Storm Topology that is based on reading Apache file and process it, I am getting weird error when submitting Topology:</p>
  
  <pre><code>6044 [main] WARN  backtype.storm.daemon.nimbus - Topology submission exception. (topology name='apachelog') #&lt;InvalidTopologyException InvalidTopologyException(msg:Component: [lineBolt] subscribes from non-existent component [line])&gt;
  6051 [main] ERROR org.apache.zookeeper.server.NIOServerCnxn - Thread Thread[main,5,main] died
  backtype.storm.generated.InvalidTopologyException: null
      at backtype.storm.daemon.common$validate_structure_BANG_.invoke(common.clj:151) ~[storm-core-0.9.0.1.jar:na]
      at backtype.storm.daemon.common$system_topology_BANG_.invoke(common.clj:287) ~[storm-core-0.9.0.1.jar:na]
      at backtype.storm.daemon.nimbus$fn__5528$exec_fn__1229__auto__$reify__5541.submitTopologyWithOpts(nimbus.clj:932) ~[storm-core-0.9.0.1.jar:na]
      at backtype.storm.daemon.nimbus$fn__5528$exec_fn__1229__auto__$reify__5541.submitTopology(nimbus.clj:950) ~[storm-core-0.9.0.1.jar:na]
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_65]
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_65]
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_65]
      at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_65]
      at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93) ~[clojure-1.4.0.jar:na]
      at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28) ~[clojure-1.4.0.jar:na]
      at backtype.storm.testing$submit_local_topology.invoke(testing.clj:236) ~[storm-core-0.9.0.1.jar:na]
      at backtype.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:19) ~[storm-core-0.9.0.1.jar:na]
      at backtype.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-0.9.0.1.jar:na]
      at storm.starter.ApacheAccessLogTopology.main(ApacheAccessLogTopology.java:68) ~[classes/:na]
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_65]
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_65]
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_65]
      at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_65]
      at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120) ~[idea_rt.jar:na]
  </code></pre>
  
  <p><strong>My Sprout file</strong></p>
  
  <pre><code>public class LogReaderSpout extends BaseRichSpout
  {
      private SpoutOutputCollector _collector;
      private String filePath = "access_log";
  
      @Override
      public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)
      {
          _collector = collector;
      }
  
      @Override
      public void nextTuple()
      {
          Utils.sleep(100);
          String line = null;
          try
          {
              BufferedReader bufferReader = new BufferedReader(new FileReader(filePath));
              while((line = bufferReader.readLine()) != null)
              {
                  if(line!=null)
                  {
                      System.out.println(line);
                      _collector.emit(new Values(line));
                  }
              }
  
          }
          catch (Exception e)
          {
              e.printStackTrace();
          }
          System.out.println("Emitting Next Tuple..");
      }
  
      @Override
      public void ack(Object id)
      {
          System.out.println("Ack with ID: "+id);
      }
  
      @Override
      public void fail(Object id)
      {
          System.out.println("Fail with ID: "+id);
      }
  
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("line"));
      }
  }
  </code></pre>
  
  <p><strong>Topology</strong></p>
  
  <pre><code>package storm.starter;
  import backtype.storm.Config;
  import backtype.storm.LocalCluster;
  import backtype.storm.StormSubmitter;
  import backtype.storm.task.OutputCollector;
  import backtype.storm.task.TopologyContext;
  import backtype.storm.topology.OutputFieldsDeclarer;
  import backtype.storm.topology.TopologyBuilder;
  import backtype.storm.topology.base.BaseRichBolt;
  import backtype.storm.tuple.Fields;
  import backtype.storm.tuple.Tuple;
  import backtype.storm.tuple.Values;
  import backtype.storm.utils.Utils;
  import storm.starter.spout.LogReaderSpout;
  
  import java.util.Map;
  
  public class ApacheAccessLogTopology
  {
      public static class LineBolt extends BaseRichBolt
      {
          OutputCollector _collector;
  
          @Override
          public void prepare(Map conf, TopologyContext context, OutputCollector collector)
          {
              _collector = collector;
          }
  
          @Override
          public void execute(Tuple tuple)
          {
              //ALL PROCESSING will take place here on tuple(in our case Line here)
              _collector.emit(tuple, new Values(tuple.getString(0) + "???"));
              _collector.ack(tuple);
          }
  
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer)
          {
              declarer.declare(new Fields("line"));
          }
  
  
      }
  
      public static void main(String[] args) throws Exception
      {
          TopologyBuilder builder = new TopologyBuilder();
          builder.setSpout("lineSpout", new LogReaderSpout(),2);
          builder.setBolt("lineBolt", new LineBolt(),2).shuffleGrouping("line");
  
          Config conf = new Config();
          conf.setDebug(true);
  
          if (args != null &amp;&amp; args.length &gt; 0) {
              conf.setNumWorkers(2);
  
              StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
          }
          else {
  
              LocalCluster cluster = new LocalCluster();
              cluster.submitTopology("apachelog", conf, builder.createTopology());
              Utils.sleep(10000);
              cluster.killTopology("apachelog");
              cluster.shutdown();
          }
      }
  
  }
  </code></pre>