Storm Topology: proper way to ack when two bolts have same source bolt

Question

I'm fairly new to Storm, and recently changed my bolts to inherit from IRichBolt instead of BaseBasicBolt, meaning I am now in charge of acking and failing a tuple according to my own logic.

My topology looks like this: Bolt A emits the same tuple to Bolts B and C, each persist data to Cassandra. These operations are not idempotent, and include an update to two different counter column families. I am only interested in failing the tuple and replaying it in certain exceptions from Cassandra (not read/write timeouts, only QueryConsistency or Validation exception). The problem is that in case bolt B fails, the same tuple is replayed from the spout and is emitted again to bolt C, which already succeeded to persist the its data, creating false data.

I've tried to understand how exactly acking is done (from reading: http://www.slideshare.net/andreaiacono/storm-44638254) but failed to understand what happens in the situation I described above.

The only way I figured to solve this correctly is to either create another spout with the same input source: Spout 1 -> bolt A -> bolt B, and Spout 1' -> Bolt A' -> Bolt C', or either to persist the data for both column family in the same Batch Statement that is done in Bolts B and C by combining them into one.

Is this correct or am I missing something? And Is there another possible solution to properly ack these tuples?

Thanks.

Answer 1

You didn't say how long you want to wait to retry an failed update in bolt B or C, but instead of outright failing the tuple in bolt B, you could add some more streams. Add a scorpion-tail output stream from bolt B back to the same bolt B. If an update in bolt B fails, write the tuple to the scorpion-tail output stream so it comes right back as input into bolt B again, just from a second stream. You could enrich the tuple to hold a timestamp so your processing logic on bolt B for the new stream could look at the last attempted time and if enough time hasn't passed you could write it out to the scorpion-tail stream again. Of course you'd do the same thing for bolt C.

If you want to wait a long time to retry the tuple (long in Storm terms), you could replace those scorpion-tail streams with Kafka topics along with the requisite spouts.