title: Mapping username to tweets with storm
body: <p>I'm trying to create a topology that has:1 spout that emits tweets and two  bolts:
   a TweetParserBolt that collects tweets 
   and UserParserBolt that collects the tweeters' username.</p>
  
  <p>Suppose I've created a third bolt that anchors the TweetParserBolt and the UserParserBolt so that it can  map the  tweeter's username to a list of tweets that she/he have already posted.The problem I've encountered is that the bolt returns a null list of tweets. </p>
  
  <p>Can anyone please help me understand what's wrong with the code </p>
  
  <p>Below is my code for the topology and the three bolts:</p>
  
  <pre><code>public class TwitterTopology {
  private static String consumerKey = "*********************";
  private static String consumerSecret = "*****************";
  private static String accessToken = "********************";
  private static String accessTokenSecret = "****************";
  
  public static void main(String [] args) throws Exception{
  
      /*** SETUP ***/
  
      String remoteClusterTopologyName = null;
      if (args!=null) {
          if (args.length==1) {
              remoteClusterTopologyName = args[0];
          }
          // If credentials are provided as commandline arguments
          else if (args.length==4) {
              accessToken =args[0];
              accessTokenSecret =args[1];
              consumerKey =args[2];
              consumerSecret =args[3];
          }
  
      }
      /**************** ****************/
  
      TopologyBuilder builder = new TopologyBuilder();
  
      FilterQuery filterQuery = new FilterQuery();
      filterQuery.track(new String[]{"#cloudcomputing"});
      filterQuery.language(new String[]{"en"});
  
      TwitterSpout spout = new TwitterSpout( accessToken, accessTokenSecret,consumerKey, consumerSecret, filterQuery);
      builder.setSpout("TwitterSpout",spout,1);
  
      builder.setBolt("TweetParserBolt",new TweetParserBolt(),4).shuffleGrouping("TwitterSpout");
      builder.setBolt("UserMapperBolt",new UserParserBolt()).shuffleGrouping("TwitterSpout");
  
      UserAndTweetsMapperBolt()).fieldsGrouping(("TweetParserBolt"), new Fields("username","tweet","bolt"))
                                                                              .fieldsGrouping(("UserMapperBolt"),new Fields("username","tweet","bolt"));
  
      Config conf = new Config();
      conf.setDebug(true);
  
  
      if (remoteClusterTopologyName!=null) {
          conf.setNumWorkers(4);
  
          StormSubmitter.submitTopology(remoteClusterTopologyName, conf, builder.createTopology());
      }
      else {
          conf.setMaxTaskParallelism(3);
  
          LocalCluster cluster = new LocalCluster();
          cluster.submitTopology("test", conf, builder.createTopology());
  
          Thread.sleep(460000);
  
          cluster.shutdown();
      }
  
  
  }
  
  public class TweetParserBolt extends BaseRichBolt {
     private OutputCollector collector;
  
     @Override
     public void declareOutputFields(OutputFieldsDeclarer declarer){
        declarer.declare(new Fields("username","tweet","bolt"));
     }
  
  
  
     @Override
     public void prepare(Map map,TopologyContext context,OutputCollector collector){
        this.collector=collector;
     }
  
     @Override
     public void execute(Tuple tuple){
        Status tweet=(Status)tuple.getValue(0);
        String username=tweet.getUser().getScreenName();
  
        collector.emit(tuple,new Values(username,tweet,"tweet_parser_bolt"));
  
      }
    }
  
     public class UserParserBolt extends BaseRichBolt{
       private OutputCollector collector;
  
       @Override
       public void declareOutputFields(OutputFieldsDeclarer declarer){
          declarer.declare(new Fields("username","tweet"));
       }
  
  
  
       @Override
       public void prepare(Map map,TopologyContext context,OutputCollector collector){
          this.collector=collector;
       }
  
       @Override
       public void execute(Tuple tuple){
         Status tweet=(Status)tuple.getValue(0);
         String username=tweet.getUser().getScreenName();
  
         collector.emit(tuple,new Values(username,tweet,"user_parser_bolt"));
  
      }
   }
  
   public class UserAndTweetsMapperBolt extends  BaseRichBolt {
      private OutputCollector collector;
  
      List&lt;Tuple&gt;listOfTuples;
  
  
      Map&lt;String,Status&gt;tempTweetsMap;
      Map&lt;String,List&lt;Status&gt;&gt;UserAndTweetsMap;
      List&lt;Status&gt;tweets;
      List&lt;String&gt;tempUsers;
  
  
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer){
         declarer.declare(new Fields("username","tweets"));
      }
  
      @Override
      public void prepare(Map map,TopologyContext context,OutputCollector collector){
        this.collector=collector;
        this.listOfTuples=new ArrayList&lt;Tuple&gt;();
  
        this.tempTweetsMap=new HashMap&lt;String, Status&gt;();
        this.UserAndTweetsMap=new HashMap&lt;String, List&lt;Status&gt;&gt;();
  
        this.tempUsers=new ArrayList&lt;String&gt;();
        this.tweets=new ArrayList&lt;Status&gt;();
      }
  
  
  
      @Override
      public void execute(Tuple tuple){
         //String username=tuple.getStringByField("username");
         //Status status=(Status)tuple.getValueByField("tweet");
  
         String username=tuple.getValue(0).toString();
  
         String sourceComponent=tuple.getSourceComponent();
  
         if(sourceComponent.equals("TwitterParserBolt")){
            String tempUser1=tuple.getValue(0).toString();
            Status tempStatus1=(Status)tuple.getValue(1);
  
            tempTweetsMap.put(tempUser1,tempStatus1);
  
         }else if(sourceComponent.equals("UserParserBolt")){
            String tempUser2=tuple.getValue(0).toString();
            Status tempStatus2=(Status)tuple.getValue(1);
  
            tempUsers.add(tempUser2);
        }
  
  
        for(int i=0;i&lt;tempUsers.size();i++){
           for(int j=0;j&lt;tempTweetsMap.size();j++){
               if(tempUsers.get(i).equals(tempTweetsMap.get(j).getUser().getScreenName())){
                   tweets.add(tempTweetsMap.get(j));
               }
           }
       }
  
  
          collector.emit(new Values(username,tweets));
     }
   }
  </code></pre>