Arroyo 0.16 is around the corner, bringing with it over a hundred changes improving the stability, performance, and scalability of the system. Ahead of the release, we’ll have a series of blog posts talking about the coming improvements.

This post will dive into a new way to run the control plane, which we call worker-leader mode. This mode enables the Arroyo data plane to run independently of the control plane, improving reliability of multi-tenant deployments, and enabling massive numbers of concurrent pipelines (at Cloudflare we’re using it to currently run thousands on a single control plane deployment for Cloudflare Pipelines).

It’s controllers all the way down

To understand the change, we’ll need to start with how an Arroyo cluster is architected.

Arroyo is broadly made up of two systems, which we call the control plane and data plane. The data plane consists of a per-pipeline cluster of workers, which actually read data off of sources, process it, and write it to sinks according to the pipeline’s SQL definition. The control plane consists of the config database, the SQL planner, REST API, Web UI, and a service which we call the controller.

The controller—so called because it works according to the Controller pattern popularized by Kubernetes—is responsible for noticing newly created pipelines, scheduling worker nodes to execute the pipeline, and then transitioning the pipeline to the steady Running state. In this steady state the controller is waiting to respond to user config changes (like stopping the pipeline) and is also monitoring the pipeline’s execution. If it encounters a fatal error (for example, because a node went down) the controller recovers it. This ensures we have continuous progress processing our input data, without needing to wake anyone up for routine issues.

This is all about the same as you’d have for any kind of service; the Kubernetes ReplicaSet controller is doing something very similar, for example. But operating a stateful streaming pipeline is a bit more involved. In particular, we have to periodically (by default, every 10 seconds) checkpoint the state of the pipeline to remote storage (typically, S3) so that we can recover from failure without losing data. Checkpointing also drives our two-phase commit (2PC) protocol which enables exactly-once processing.

The checkpointing algorithm Arroyo uses (based on Asynchronous Barrier Snapshotting) requires someone to instruct the source nodes in the operator graph to initiate checkpointing. Since Arroyo’s first version, that role has belonged to the controller. (In this, we took inspiration from the Flink system architecture, which uses its JobManager similarly). But there’s a key problem here: the controller has to be up and running, or we won’t checkpoint or commit.

There’s a second, dumber issue. Our 2PC protocol guarantees that processing, given a single Arroyo cluster, is exactly once. But if we somehow are running two clusters for the same pipeline¹ … we’re going to double write anyway. Something has to ensure that we don’t end up with multiple running clusters. Historically, we’ve relied on an active heartbeat from the workers to the controller. If the controller became unreachable, or informed the worker that there was a new cluster for the pipeline, the workers would terminate². Unfortunately, this means that the controller going down takes down all pipelines—unacceptable for a large, multi-tenant service.

So at Cloudflare, we ended up running Arroyo pretty much how everyone runs Flink: with a single-tenant controller per pipeline cluster. Unfortunately this is quite expensive, particularly for the long tail of small pipelines, and makes operating all of these clusters very challenging. So we set out to fix this, and make Arroyo truly suitable for huge multi-tenant deployments.

Introducing worker leaders

The result is worker-leader mode. It’s pretty much what it sounds like! We elect a single worker in the pipeline cluster to be the cluster leader, which is responsible for orchestrating the checkpointing and 2PC processes. It also takes on other per-worker responsibilities from the controller: it receives heartbeats from each worker, collects metrics, and manages pipeline startup and shutdown.

The controller is still responsible for handling and recovering from failure. We still need something outside of the cluster to make sure the cluster is working. But instead of directly receiving heartbeats from all of the workers (creating a dependency from the data plane to the control plane) it merely polls the leader periodically. The leader has its own job state machine and only needs to report the cluster state to the controller.

For example, if a cluster worker dies, it will stop sending heartbeats. After a configurable interval this will cause the leader to transition the pipeline to a failed state. The controller polls this from the leader, and triggers a recovery (tearing down the existing cluster and starting a new one). If we can’t reach the leader for a configurable allowed failure time, we similarly treat the cluster as failed and recover it.

This hierarchical approach to metrics, state, and progress inherently improves the scalability of the controller, which now scales out based on the number of pipelines, not the number of workers or subtasks. Another advantage of making all pipelines—regardless of size—uniform in terms of controller resource utilization comes about when we need to horizontally scale controllers. If all pipelines are uniform we can just randomly shard them across controllers and get even resource utilization across the entire cluster.

Exactly once

So this solves the first problem: dependency of the data plane on the control plane. But we had a second, more critical one: somehow we need to prevent split-brain issues causing multiple clusters to commit conflicting data externally. We decided to take the time to do it correctly, by designing an object storage-based protocol that ensures that for each epoch, at most one checkpoint becomes canonical and eligible for committing externally.

The full protocol description is too involved to cover in this blog post³, but I’ll give a quick sketch. The protocol is built around what we call state lineages, which are paths through the tree of checkpoints belonging to a pipeline. When a pipeline is first created, there is an initial cluster that is started. We’ll call this first cluster generation 1, or G1. The controller starts by updating a current-generation file for the job. Then on startup, the leader will write a generation manifest, marking the existence of this generation and recording the checkpoint information that we’re recovering from (since this is the first generation, that’s empty).

Each checkpoint is given an incrementing integer id, called an epoch, so each checkpoint can be uniquely identified by the triple (job, generation, epoch). On each checkpoint, after writing out the state data⁴, we write a checkpoint manifest file at a path scoped by that triple. The checkpoint manifest contains general metadata about the checkpoint as well as metadata for each operator, including references to the underlying data files⁵. At that point, we’ve completed the checkpointing phase.

Next is committing. This is the point where we synchronize our state with external transactional systems, like Kafka topics or Iceberg tables. In other words, we’ve written, for example, Parquet files into object storage in an Iceberg data directory as part of processing incoming data since the last checkpoint, but we haven’t yet committed it to the Iceberg table, so it’s not yet part of the table or queryable by users.

We have two options: we can roll back or abandon our transaction (containing the updates since the last checkpoint) or we can commit it. To avoid double writes, we need to ensure that exactly one checkpoint will be committed for this epoch. We accomplish this via a conditional write (using If-None-Match⁶), this one to an epoch-scoped path, which we call the epoch record. The epoch record marks the canonical generation-scoped checkpoint for a particular epoch.

This is a bit subtle: an epoch record with generation G and epoch E says that the specific checkpoint that must be committed is the checkpoint at path (G, E) — not that the generation G cluster must be the one to commit it. But it does rule out committers. If we have just checkpointed (G’, E) and lost the race to write the epoch record, then we are no longer tracking the true lineage and must shut down.

At some future point, G1 dies, and a new one (G2) is started. On startup, we’ll find the last generation’s manifest (in this case G1) and read the last canonical checkpoint. If the checkpoint was not marked as committed, we’ll redo the commit⁷ (first verifying that we’ve recovered from the checkpoint that owns the epoch record), and continue on from that point.

The fundamental invariant we must maintain is: there is only one true lineage for the pipeline, and only checkpoints on that lineage are committed. For example, here’s a checkpoint history across three generations representing a split-brain scenario. We start G1 and it checkpoints epochs 1 and 2. Then, the controller loses connection to the leader, and we start a new cluster, G2, which restores checkpoint (G1, 2) and starts to checkpoint epoch 3. Unfortunately, G1 is still running, and also checkpoints epoch 3. At this point either G1 or G2 could win the race to write the epoch 3 record. In this case, G1 wins it, and G2 shuts down. This causes the controller (which thinks G2 is the current cluster for this job) to shut it down and schedule a new G3, which now owns the lineage⁸:

Coming up

Worker leaders are available now in the Arroyo master branch (released as the :tip Docker image) and can be enabled by configuring the job-controller: worker config setting (ARROYO__JOB_CONTROLLER=worker as an env-var). At Cloudflare, we’ve been running this mode in production across thousands of pipelines for several months, but as this represents a huge change to a fundamental part of Arroyo, we are opting to disable it by default in the upcoming Arroyo 0.16.0 release. In Arroyo 0.17.0, the next minor release, we will remove the existing mode (which we call controller) and worker-leader mode will be enabled for all clusters.

Beyond the scalability, reliability, and correctness improvements detailed here, there’s another reason we’re excited about this change. In controller mode, much of the complex behavior around executing pipelines lived in the control plane. But with the worker cluster in the data plane now largely self-sufficient once it’s up and running, this will enable users to build their own control planes that are specialized to their environments and use cases.