Storm topology hangs up

[Dimitris Sarlis <sa...@gmail.com>]

Hello,

I'm trying to run a topology containing some spouts that take input from 
a Kafka queue and redirect that input to a pool of bolts. After just a 
few records have been sent, the topology seems to hang up and no more 
messages can be digested. I've set the TOPOLOGY_MAX_SPOUT_PENDING to 500 
and approximately, when the first 500 messages are sent, the bolts stop 
processing other records. However, through Ganglia I can see that each 
worker has a CPU utilization of ~20% even though it seems that bolts 
don't perform any processing.

My configuration is as follows: 1 master running the nimbus and 14 
slaves running 4 supervisor slots each. I deploy 14 spouts and 42 bolts 
using the following code:

for (int i = 1; i <= boltNo; i ++) {
       bd = builder.setBolt("worker" + i, new ExclamationBolt(boltNo), 
boltPar);
       for (int j = 1; j <= spoutNo; j++)
             bd = bd.shuffleGrouping("words" + j);
       for (int k = 1; k <= boltNo; k++) {
             bd = bd.directGrouping("worker" + k);
      }
}

Essentially, spouts shuffle records to bolts and each bolt is connected 
with a direct grouping with the other bolts.

The execute method of my ExclamationBolt is:

public void execute(Tuple tuple) {
         if (!tuple.getString(0).contains("!")) {
             Random ran = new Random();
             int worker = ran.nextInt(boltNo) + 1;
             List<Integer> l = _topo.getComponentTasks("worker" + worker);
             LOG.info("List: " + l + " for worker: " + worker);
             String out = tuple.getString(0) + "!";
             LOG.info("Sending record to appropriate worker: worker" + 
worker);
             _collector.emitDirect(l.get(0), new Values(out));
         }
         else {
             LOG.info("Already processed this record");
         }
         _collector.ack(tuple);
}

As you can see, if the record does not contain a "!", I append a "!" and 
send it to a random worker (performing one hop) or else I just print a 
log message and proceed to the next record.

Any ideas what might be wrong with this scenario?

Thanks,
Dimitris Sarlis