High Availability of Computation Tasks

To ensure the stable execution of stream computation tasks in a distributed environment, Orca employs a stream-graph-level Checkpoint mechanism. This mechanism enables fast task recovery in the event of node failures, network interruptions, or storage errors, ensuring no data loss.

Checkpoint Mechanism

Orca's Checkpoint implementation is based on the Chandy–Lamport distributed snapshot algorithm, which introduces Barrier markers to define consistency boundaries within the data flow, thus producing globally consistent snapshots.

The core Checkpoint process includes the following steps:

  • The Checkpoint Coordinator periodically triggers Checkpoint tasks and injects Barrier markers into all source nodes.
  • Upon receiving a Barrier, the source node atomically writes it to the stream table and records the current data offset.
  • As the Barrier flows through the stream graph, each engine or stream table that receives it from upstream performs a local state snapshot, and then passes the Barrier downstream.
  • When a sink node receives the Barrier, it indicates that all its upstream nodes have completed their snapshots. Once all sink nodes receive the Barrier, the Checkpoint is considered complete.

End-to-End Consistency

There are two levels of consistency semantics:

  • AT_LEAST_ONCE: Data is processed at least once—no data is lost, but duplicates may occur.
  • EXACTLY_ONCE: Data is processed exactly once—no loss, no duplicates.

Orca provides end-to-end consistency—from source nodes to sink nodes—by implementing the following guarantees:

  • Source-side consistency is achieved by persisting stream tables and tracking data offsets, enabling replay from a known position upon restart.
  • Computation task consistency is guaranteed by the Chandy–Lamport snapshot algorithm, ensuring globally consistent snapshots across the distributed system.
  • Sink-side consistency is determined by the type of stream table used. EXACTLY_ONCE is only achievable if all public stream tables (except sources) are either keyedStreamTable or latestKeyedStreamTable. These tables provide deduplication, filtering out redundant upstream data.
Note:

EXACTLY_ONCE semantics rely on in-memory key-level deduplication. Once a record with a given key is persisted, the system may still accept a new record with the same key.

Therefore, the system ensures uniqueness in memory, not globally. This is sufficient to handle multi-writer or delayed submission scenarios that may lead to duplicates.

Barrier Alignment

For EXACTLY_ONCE consistency, Barrier alignment is required when there are multiple upstream paths. A stream task must wait for all upstream Barriers before taking a snapshot. This process is known as Barrier alignment.

To achieve this, Orca introduces a lightweight intermediate component called Channel, inserted into every stream task except source nodes, with the following behavior:

  • Each Channel is placed on a distinct upstream path to an engine or stream table. When a Barrier is received, the Channel pauses data transmission on that path.
  • Once all Channels receive the Barrier, the stream task executes local snapshots of all downstream engines or stream tables in order.
  • After snapshot completion, the Barrier is forwarded to the output stream table and subsequently to downstream tasks.