In the final section of this topic, let’s consider how stream processors can tolerate faults. We saw in the related discussion that batch processing frameworks can tolerate faults fairly easily: if a task fails, it can simply be started again on another machine, and the output of the failed task is discarded. This transparent retry is possible because input files are immutable, each task writes its output to a separate file, and output is made visible only when a task completes successfully.
In particular, the batch approach to fault tolerance ensures that the output of the batch job is the same as if nothing had gone wrong, even if some tasks did fail. It appears as though every input record was processed exactly once—no records are skipped, and none are processed twice. Although restarting tasks means that records may be processed multiple times, the visible effect in the output is as if they had been processed only once. This principle is known as exactly-once semantics, although effectively-once would be a more descriptive term.
The same issue of fault tolerance arises in stream processing, but it is less straightforward to handle. Waiting until a task is finished before making its output visible is not an option, because a stream is infinite, so you can never finish processing it.
Microbatching and checkpointing
One solution is to break the stream into small blocks and treat each block like a miniature batch process. This approach is called microbatching, and it is used in Spark Streaming. The batch size is typically around one second, which is the result of a performance compromise: smaller batches incur greater scheduling and coordination overhead, while larger batches mean a longer delay before results of the stream processor become visible.
Microbatching also implicitly provides a tumbling window equal to the batch size (windowed by processing time, not event timestamps); any jobs that require larger windows need to explicitly carry over state from one microbatch to the next.
A variant approach, used in Apache Flink, is to periodically generate rolling checkpoints of state and write them to durable storage. If a stream operator crashes, it can restart from its most recent checkpoint and discard any output generated between the last checkpoint and the crash. The checkpoints are triggered by barriers in the message stream, similar to the boundaries between microbatches, but without forcing a particular window size.
Within the confines of the stream processing framework, the microbatching and checkpointing approaches provide the same exactly-once semantics as batch processing. However, as soon as output leaves the stream processor (e.g., when it writes to a database, publishes messages to an external message broker, or triggers the sending of emails), the framework is no longer able to discard the output of a failed microbatch. In this case, restarting a failed task causes the external side effect to happen twice, and microbatching or checkpointing alone is not sufficient to prevent this problem.
Atomic commit revisited
To give the appearance of exactly-once processing in the presence of faults, we need to ensure that all outputs and side effects of processing an event persist if and only if the processing is successful. That includes any messages sent to downstream operators or external messaging systems (including email or push notifications), database writes, changes to operator state, and acknowledgments of input messages (including moving the consumer offset forward in a log-based message broker).
These actions must all happen atomically: either they all occur, or none of them do. If this approach sounds familiar, that’s because we discussed it in “Exactly-once message processing” in the context of distributed transactions and two-phase commit.
We considered the problems with the traditional implementations of distributed transactions, such as XA, in the related discussion. However, in more restricted environments it is possible to implement such an atomic commit facility efficiently. This approach is used in Google Cloud Dataflow, VoltDB, and Apache Kafka. Unlike XA, these implementations do not attempt to provide transactions across heterogeneous technologies, but instead keep the transactions internal by managing both state changes and messaging within the stream processing framework. The overhead of the transaction protocol can be amortized by processing several input messages within a single transaction.
Idempotence
Our goal is to discard the partial output of any failed tasks so that they can be safely retried. Distributed transactions are one way of achieving that goal; another is to rely on idempotence, as we saw in “Durable Execution and Workflows”.
An idempotent operation is one that you can perform multiple times, and it has the same effect as if you performed it only once. For example, deleting a key in a key-value store is idempotent (deleting the value again has no further effect), whereas incrementing a counter is not idempotent (performing the increment again means the value is incremented twice).
Even if an operation is not naturally idempotent, it can often be made idempotent with a bit of extra metadata. For example, when consuming messages from Kafka, every message has a persistent, monotonically increasing offset. When writing a value to an external database, you can include the offset of the message that triggered the last write with the value. Thus, you can tell whether an update has already been applied and avoid performing the same update again. The state handling in Storm’s Trident is based on a similar idea.
Relying on idempotence implies several assumptions: restarting a failed task must replay the same messages in the same order (a log-based message broker does this), the processing must be deterministic, and no other node may concurrently update the same value. When failing over from one processing node to another, fencing may be required (see “Distributed Locks and Leases”) to prevent interference from a node that is thought to be dead but is actually alive. Despite all those caveats, idempotent operations can be an effective way of achieving exactly-once semantics with only a small overhead.
Rebuilding state after a failure
Any stream process that requires state—for example, windowed aggregations (such as counters, averages, and histograms) and any tables and indexes used for joins—must ensure that this state can be recovered after a failure.
One option is to keep the state in a remote datastore and replicate it, although having to query a remote database for each individual message can be slow. An alternative is to keep state local to the stream processor and replicate it periodically. Then, when the stream processor is recovering from a failure, the new task can read the replicated state and resume processing without data loss.
For example, Flink periodically captures snapshots of operator state and writes them to durable storage such as a distributed filesystem, and Kafka Streams replicates state changes by sending them to a dedicated Kafka topic with log compaction, similar to CDC. VoltDB replicates state by redundantly processing each input message on several nodes (see “Actual Serial Execution”).
In some cases, replicating the state may not even be necessary, because it can be rebuilt from the input streams. For example, if the state consists of aggregations over a fairly short window, it may be fast enough to simply replay the input events corresponding to that window. If the state is a local replica of a database, maintained by CDC, the database can also be rebuilt from the log-compacted change stream.
All this depends on the performance characteristics of the underlying infrastructure. In some systems, network delay may be lower than disk access latency, and network bandwidth may be comparable to disk bandwidth. No solution is universally ideal for all situations, and the merits of local versus remote state may also shift as storage and networking technologies evolve.