So far, we have understood very well the function of the nextTuple() method in the spout. It fetches the next event that becomes available at the source and emits it into the topology. The open() method of the spout holds the definition of the spout collector that's used to actually emit the tuple from the spout into the topology. Every tuple/event that is emitted by the spout into the topology is tagged with a message ID, which is the identifier for the tuple. Whenever the spouts emit the messages into the topology, it tags them with messageId, as shown here:

_collector.emit(new Values(...),msgId);

This messageId tag is actually an identifier for the tuple. Storm tracks and tags the tuples using this messageID as it plays the tuple through the bolt tree.

In case the event is successfully played through the DAG, it's acknowledged. Please note that this acknowledging of the tuple is done at the spout level by the spout that emitted the tuple. If the tuple is timed out, then the originating spout executes a fail() method on the tuple.

Now, we need to understand a very important aspect of replay. The question is, "how does Storm replay the tuples that are already emitted?" The answer is that the spout reads a tuple from the queue, but the tuple remains in the queue until it's successfully played and then once ack() is received only then does Storm acknowledge the tuple to the queue and thus removes the same. In case of fail(), the message is queued back to the queue for consumption and replay:

As depicted in the diagram, the failed tuples are propagated back to the spout and are lined back into the queue. When the messages are read from the queue by the spout, they are marked as un-acked. During this duration, they are still in the queue but not available to be read by any other spout/topology/client. If such a un-acked message is played successfully through the topology it's acked and removed from the queue. In case the event fails to complete its execution through the topology, it's failed and re-queued and made available for consumption to the spout again.

One of the most important points to note is anchoring. It's tagging of all edges of the DAG through which the tuple is played. Anchoring is a choice of the developer who wires in the topology, but Storm can replay the messages only in anchored topologies. Now the two logical questions to be asked are:

Let's answer these one by one. Anchoring is pretty simple and we as developers have to be cognizant of the same during emitting the tuples from the bolts and spouts. Here are two code snippets that would make it very clear:

_collector.emit(new Values(word));

The above snippet is the un-anchored version from unreliable topologies, where the tuples could not be replayed in the event of failure.

//simple anchoring
_collector.emit(new Values(1, 2, 3),msgID);
List<Tuple> anchors = new ArrayList<Tuple>();
anchors.add(tuple1);
anchors.add(tuple2);
//multiple anchoring output tuple is anchored to two //input tuples viz tuple1 and tuple2
_collector.emit(anchors, new Values(1, 2, 3));

This second snippet captures the anchored version, where the tuples are tied together using the anchor list, and in case of any failure, this reliable topology would be able to replay all the events.

Now coming to our second question, having anchoring and reliable topology is a little expensive in terms of book keeping and message size that's being propagated through the topology. There could be numerous scenarios where reliability is not required and hence there are unreliable topologies where no anchoring is done.

Storm bolts can be broadly classified into two segments: the basic bolts and the ones that do the aggregation and joins. The basic bolts are plain and clearly depicted in the following diagram; prepare() sets up the output collector to which tuples are emitted immediately after they are processed. All the bolts following this general pattern are actually implemented using the IBasicBolt interface. The diagram also captures the second variety of not-so-simple Storm bolts that perform the tasks such as aggregation and joins. The tuples are anchored to multiple input tuples and sometimes they are emitted after a preprogrammed delay such as in the case of aggregates:

Now that we have understood Storm reliability, let's touch upon the most important component that makes sure that all updates about ack or fail reach the spout successfully. They are acker processes.

Now that we have understood Storm reliability, let's discuss the most important component that makes sure that all updates about ack or fail reach the spout successfully. They are acker processes. These are lightweight tasks that coexist with spouts and bolts, and are responsible for acking all successfully executed messages to the spout. There, the number is controlled by a property in Storm config called TOPLOGY_ACKER_EXECUTORS; by default, it's equal to the number of workers defined in the topology. If the topology operates at a less TPS, the num ackers should be reduced. However, for high-TPS topologies, this number should be increased so that tuples are acknowledged at the rate of arrival and processing.

Ackers follow the following algorithm to track the completion of the tuple tree for each tuple being played through the topology:

If all tuples in the tuple tree are acked the checksum would be 0; or else, it would be a non-zero value, the latter denoting a failure in the topology. Its working is largely driven and controlled using tick tuple:

The preceding diagram captures the flow of the acker process. Note that it has a book keeping component, ledger or rotatingMap, and the entire flow of things is controlled on the basis of tick tuples.