Source-linked AI summary
Koala: Non-disruptive Reconfiguration for Long-running Stateful Dataflows
Lei Huang, Sakshi Sharma, Chenxu Yu, Zikun Wang, Yuanli Wang, Zichen Zhu, John Liagouris, Vasiliki Kalavri
TL;DR
Long-running dataflows must balance low-latency embedded state against the disruption of moving it during reconfiguration. Koala lazily migrates needed state using lightweight ownership transfer; it recovers backlog up to 23× faster than baselines and transfers 8.7× less state on average. Its reported evaluation also identifies limits in the experimental settings used by prior work.
Problem
Embedded state offers low steady-state latency but must move during reconfiguration, while external state avoids that migration but incurs higher latency and resource consumption.
Source: §1 · p. 1, §1.1 · p. 2Method
Results
23× faster backlog recovery than baselines and 8.7× less state transferred on average.
Source: §1.2 · p. 3Takeaways & Limitations
Koala bridges embedded-state performance and remote-state flexibility with non-disruptive fine-grained reconfiguration.
Source: §7 · p. 12Prior evaluations often use small state, stable backpressure-free inputs, or single-operator reconfiguration rather than production-like conditions.
Source: §1.1 · p. 2
Abstract
from arXiv · showhide
Stateful dataflow systems must decide where state lives relative to the tasks that process it. For long-running applications that require frequent reconfiguration due to changes in resource allocation, partitioning, or task placement, this design choice presents a core trade-off between adaptability and performance. Co-locating state with computation ensures low latency at steady state but makes reconfiguration disruptive, as any change forces state migration. On the other hand, keeping state in an external store makes reconfiguration seamless, but incurs overhead on every access. We present Koala, a reconfiguration protocol for distributed dataflow systems that retains the operational performance of embedded-state architectures while matching the disruption-free adaptability of external-state designs. Unlike prior protocols which relocate all state regardless of whether it will soon be accessed, Koala migrates state in a fine-grained, lazy manner. To achieve that, it employs a lightweight ownership transfer mechanism that safely decouples event routing from state location. We implement Koala in a new streaming dataflow system and evaluate it on real-world and synthetic benchmarks. When compared with the state-of-the-art protocol DRRS, Koala consumes backlog up to 23x faster and transfers 8.7x less state on average during reconfiguration.
1 Introduction
Koala targets the trade-off between embedded state’s low steady-state latency and external state’s disruption-free reconfiguration. It lazily migrates only needed state, addressing limitations in prior live protocols, and reports faster backlog recovery with less state transfer.
1 Introduction
- Koala’s rescaling experiment transfers 6× less state, has no disruption, and matches embedded-state performance while remote state incurs 1.5× higher processing latency.The stateful operator’s parallelism is updated after three minutes; Koala fetches hot keys on demand.Source: §1 · p. 1
1.1 Limitations of existing live approaches
- Prior live protocols still migrate entire partitions, repeatedly moving cold state under diurnal or bursty workloads.Asynchronous or progressive migration still consumes resources and can disrupt performance.Source: §1.1 · p. 2
- Prior protocols often leave event-time semantics underspecified, support restricted operators, or address only limited scenarios such as scale-out.These gaps complicate adoption in real systems.Source: §1.1 · p. 2
- Prior evaluations often use small state, stable backpressure-free inputs, or single-operator changes, unlike production settings described in the paper.Production workloads may involve an order-of-magnitude larger state, overload-related backlog, and concurrent changes to the same dataflow.Source: §1.1 · p. 2
1.2 Technical challenges and contributions
- Koala must ensure correctness despite divergent routing and state location, track state locations across repeated reconfigurations, and support disruption-free full migration when required.Its contributions include lightweight ownership transfer, compact location metadata, watermark inheritance, and piggybacking additional state on remote requests.Source: §1.2 · p. 2
- 23× faster backlog recovery and 8.7× less state transferred on average than the baselines.The evaluation includes real-world and synthetic benchmarks and state sizes up to 320GB.Source: §1.2 · p. 3
2 Stateful dataflow reconfiguration
A stateful dataflow application is a graph of operators and channels, with tasks distributed across workers and keyed state assigned to tasks. Reconfiguration changes key assignment, task parallelism, or task location.
2.1 The stateful dataflow model
- A dataflow is a directed acyclic graph of operators and channels; operators run as distributed data-parallel tasks, and stateful outputs may depend on prior inputs.Each record carries a key, value, and event timestamp.Source: §2.1 · p. 3
- In embedded-state systems, each task owns one disjoint key partition and stores its key-value state locally; per-key reads and writes belong to exactly one task at a time.Tasks emit watermarks to express event-time progress, whose monotonicity is essential for time-dependent computation.Source: §2.1 · p. 3
2.2 Reconfiguration scenarios
- Reconfiguration can add or remove tasks, rebalance keys among existing tasks, or move a task to another node.Scale-out and scale-in change parallelism; key rebalance does not, while task migration changes physical location without changing key partitioning.Source: §2.2 · p. 3
3 The Koala reconfiguration protocol
Koala decouples key ownership from physical state location, so reconfiguration can take effect before state is migrated. It fetches state on demand, while preserving event-time semantics and supporting repeated reconfigurations and common operator and repartitioning patterns.
3 The Koala reconfiguration protocol
- Koala transfers ownership metadata separately from state, letting new owners fetch remote state on demand while old owners retain it until access.Keys that are never accessed incur no migration overhead.Source: §3 · pp. 3–4
3.1 Protocol overview
- During scale-out, senders transfer KLT entries before upstream tasks update routing, then forward records for reassigned keys to the receivers.Receivers fetch state on first access; subsequent accesses are local, while untouched keys remain at the old owner.Source: §3.1 · pp. 4–5, Figure 3 · p. 4
3.2 Watermark propagation
- Koala initializes new tasks from sender watermarks and aligns event time using the minimum watermark across upstream and peer channels.Receivers can process records while waiting for watermark messages; event-time progress governs when they can safely emit time-based results.Source: §3.2 · pp. 5–6, Figure 4 · p. 5
3.3 Supporting multiple reconfigurations
- KLT entries passed at each ownership transfer let owners fetch state in one hop regardless of prior reconfigurations.The one-hop guarantee also reduces state transferred when tasks are canceled because unfetched state can remain at its existing location.Source: §3.3 · p. 6
3.4 Forcing a full state transfer
- When tasks are canceled, Koala gradually migrates their remaining state by piggybacking extra keys on fetches, with N trading processing disruption against migration duration.The extra-key transfer runs in the background; unlike Megaphone’s fluid strategy [26] and DRRS’s subscales [41], it does not divide a reconfiguration or prolong ownership transfer.Source: §3.4 · pp. 6–7
3.5 Protocol generality
- Koala supports continuous aggregations, windows, joins, scale-out, scale-in, rebalance, and task migration, and is agnostic to partitioning policy and access pattern.For non-time-dependent operators it skips watermark forwarding and event-time computation; it operates correctly even under uniform access across the key space.Source: §3.5 · p. 7
4 Implementation
Koala is implemented in a new dataflow runtime with pluggable reconfiguration protocols and shared control-plane utilities. Its implementation uses configurable key-value backends, bucket-granularity lookup tables, and checkpointed KLT entries.
Implementation
- The new Go runtime supports pluggable reconfiguration protocols, enabling comparisons on a common runtime.Shared utilities provide APIs for partitioning, worker counts, placement, control messages, marker injection, routing updates, and acknowledgments.Source: §4 · p. 7
- Tasks access state through a unified backend API, with embedded or external key-value stores both backed by PebbleDB.Per-batch deduplication fetches each unique key at most once per batch and reduces serialization overhead.Source: §4 · p. 7
- The runtime checkpoints KLT entries with task snapshots so tasks can recover from the last checkpoint after failure.State backend flushing and KLT updates run in the background, with synchronization enforced only at batch boundaries.Source: §4 · pp. 7–8
- Koala maintains KLT mappings at bucket granularity and transfers a bucket’s lookup entries as a unit when ownership changes.The runtime uses 128 buckets with consistent hashing by default, while allowing deployment-specific partitioning strategies.Source: §4 · pp. 7–8
5 Evaluation
Evaluation compares Koala with prior protocols on large-state scale-out, then tests production-scale and varied reconfiguration scenarios and controlled state-size and workload factors.
5.1 Experimental methodology
- Experiments use a shared runtime for Koala and baselines, real-world and Nexmark workloads, and report throughput, Kafka lag, per-batch latency, and state-transfer metrics.The evaluation also covers backlog recovery, large-state scale-out, repeated reconfiguration, skew mitigation, task migration, and controlled microbenchmarks.Source: §5 · p. 8, §5.1 · p. 8, §5.1.1 · p. 8
5.2 Comparison with baselines on large state
- 23.2×: DRRS takes up to 23.2× longer than Koala to recover backlog; Koala transfers 8.7× less state than DRRS on average.Across queries, Koala clears backlog in 18–52 seconds, with Borg an outlier at 127 seconds; state-transfer averages are across real-world queries.Source: Figure 6 · p. 9, Figure 7 · p. 9, §5.2 · p. 10
- Koala has lower per-batch latency than Remote across all queries, averaging 34% lower while remaining stable during reconfiguration.Remote stores state externally and has no disruption during reconfiguration.Source: Figure 8 · p. 10, §5.2 · p. 10
5.3.1 Large-scale migration
- At ≈320 GB across 32 tasks, Koala’s recovery duration is comparable to §5.2 despite 10× more state and 4× more tasks.The reported explanation is that Koala migrates active keys only, making migration cost independent of task count and total state size.Source: Figure 9 · p. 10, §5.3.1 · p. 10
5.3.2 Repeated reconfigurations
- Across repeated scale-outs and scale-ins, Koala transfers 2.6 GB and 8.9 GB at two scale-in steps, versus 10 GB and 14 GB expected with full migration.Cold state is not migrated to new tasks during earlier scale-outs, so it need not be moved back from canceled tasks.Source: Figure 10 · p. 11, §5.3.2 · p. 11
5.3.3 Skew mitigation
- After a skew-aware rebalance, Koala resolves the Taxi bottleneck and converges to stable, balanced throughput from a starting point where one task exceeds 5,000 records/s.The other tasks were nearly idle before reconfiguration.Source: Figure 11 · p. 11, §5.3.3 · p. 11
5.3.4 Task migration
- For full 10 GB task migration, progressive-default sustains input rate and completes in about 5 minutes; progressive-large halves that time with a slight throughput drop.This illustrates the tunable trade-off between migration speed and disruption.Source: Figure 12 · p. 11, §5.3.4 · p. 11
5.4 Microbenchmarks
- Across runs with varying state size, Koala migrates around 1.43 GB; key lookup takes less than 40 µs per input batch, accounting for under 3% of per-batch latency.After reconfiguration, KLT memory reaches at most 60 MB for 40M keys and 80 GB of total state.Source: §5.4 · p. 12
6 Related work
Koala builds on prior techniques for fluid migration, fine-grained state transfer, and record rerouting, but distinguishes itself through active-state migration and non-blocking processing.
6 Related work
- Koala migrates only active state and avoids blocking record processing, whereas prior protocols use buffering, partial blocking, or synchronization barriers.Its markers signal routing changes and peer-channel closure rather than serving as synchronization barriers.Source: §6 · p. 12
- Koala targets long-running tasks but could also integrate with dynamic scheduling; it complements policies that decide when and how to reconfigure.Those policies do not address how reconfiguration is executed.Source: §6 · p. 12
7 Conclusion
Koala is a live reconfiguration protocol that bridges embedded-state performance and remote-state flexibility through lazy migration and decoupled key partitioning and state location.
7 Conclusion
- Koala uses lazy state migration and decouples key partition assignment from state location to support non-disruptive, fine-grained reconfiguration.The paper demonstrates it in repeated rescaling, skew mitigation, and task migration scenarios.Source: §7 · p. 12
A Protocol analysis
The appendix formalizes deployments and reconfigurations, then defines correctness through per-key isolation, exactly-once processing, and watermark correctness. These properties make reconfigured output equivalent to output under the final deployment from the start.
A Protocol analysis
- A deployment pairs each operator’s key partitioning with a mapping from its tasks to physical workers.A reconfiguration changes the target operator’s partitioning, worker mapping, or both.Source: A Protocol analysis · p. 16
A.1 Correctness properties
- Correct reconfiguration requires per-key read/write isolation, exactly-once processing, and watermark correctness throughout the transition.Watermarks must be non-decreasing and must not pass records whose event timestamps are at or before the watermark.Source: §A.1 · p. 16
- These properties ensure reconfigured output is equivalent to output under the final configuration from the start.Source: §A.1 · p. 16
A.2 Proof sketches
The proof sketches argue that Koala preserves single ownership, exactly-once processing, and watermark correctness during reconfiguration. They rely on ordered ownership handoffs, routing updates, and watermark advancement rules.
A.2 Proof sketches
- Koala preserves one responsible task per key by completing sender work, transferring ownership metadata, then redirecting subsequent records.The sender cannot access transferred state afterward; the receiver pulls it if it remains physically at the sender.Source: §A.2 · p. 16
- RT updates occur after the ownership handoff, so reassigned keys are routed to the receiver only after the sender relinquishes responsibility.Source: §A.2 · p. 16
- Exactly-once processing depends on preventing both missed records and duplicates; atomic RT updates route records exclusively over the new path.Koala does not handle failures during reconfiguration; an external checkpointing mechanism rolls back to the latest checkpoint and restarts the protocol.Source: §A.2 · pp. 16–17
- The proof sketches conclude that every record arriving at the target operator is processed exactly once.Source: §A.2 · p. 17
- The receiver withholds event-time advancement until expected markers arrive, then advances only to the minimum event time across active input channels.If a channel has not yet produced a watermark, the receiver skips the progress update.Source: §A.2 · p. 17
- Channel completeness and the minimum-time bound support monotonic watermarks and prevent advancement past records still in transit.When a peer channel closes, the sender flushes in-flight records before closure, so recomputing over remaining channels is safe.Source: §A.2 · p. 17
B Comparison with DRRS on Flink
In a comparison on Flink, DRRS-ours sustains input without disruption after reconfiguration, while DRRS-original sees a throughput drop and Kafka lag. The appendix also describes the real-world and Nexmark workloads used in evaluation.
B Comparison with DRRS on Flink
- The Flink comparison uses the Twitch query and scales ViewerLoyaltyCalculator from 4 to 8 tasks after 3 minutes in a 10-minute run.DRRS-original uses even task placement and default configuration parameters for both approaches.Source: B Comparison with DRRS on Flink · p. 17
- DRRS-ours sustains input without disruption, while DRRS-original has a slight throughput drop and Kafka lag up to 10 seconds.The figure compares throughput and Kafka lag; DRRS-original is run on Flink, while DRRS-ours is also evaluated at the runtime’s maximum processing capacity.Source: B Comparison with DRRS on Flink · pp. 17–18, Figure 16 · p. 18
C.1 Real-world queries / C.2 Nexmark queries
- The evaluation describes four real-world analytics queries and eight Nexmark queries or variants, spanning streaming engagement, taxi trends, resource monitoring, joins, and windows.Source: §C.1 · p. 18, §C.2 · p. 18
D.1 Comparison with embedded state
Koala matches the optimized stop-and-restart baseline’s steady-state latency without reconfiguration spikes. Against the Remote baseline, it has similar non-disruptive performance and higher throughput across queries.
D.1 Comparison with embedded state
- Koala achieves the same steady-state per-batch latency as S&R and shows no spikes upon reconfiguration.DRRS is omitted because per-batch latency is unavailable for that baseline.Source: §D.1 · pp. 18–19, Figure 17 · p. 19
D.2 Comparison with remote state
- Koala has similar non-disruptive performance to Remote, with small Kafka lag, while sustaining higher throughput across all queries.The passage attributes the throughput difference to low-latency state access.Source: §D.2 · p. 19, Figure 18 · p. 19
D.3 Full results on Nexmark benchmark
Across small-state Nexmark queries, Koala and the baselines perform similarly; it also sustains input under uniform access and overlaps multi-operator reconfigurations with the same backlog recovery time as a single-operator reconfiguration.
D.3 Full results on Nexmark benchmark
- Across the included small-state Nexmark queries, Koala and the baselines had similar performance and low Kafka lag; S&R dipped in throughput but recovered backlog fastest.Figures 19–20 cover Nexmark queries except Q3 and Q6*.Source: §D.3 · p. 19, Figure 19 · p. 20, Figure 20 · p. 20
D.4 Multi-operator reconfiguration
- Koala effectively overlapped two reconfigurations, matching single-operator backlog recovery through sequential protocol invocations with parallel on-demand state transfers.The Twitch query's two operators had accumulated 11.2GB and 5.6GB of state, respectively.Source: Figure 21 · p. 21, §D.4 · p. 21
D.5 Uniform state access
- About 45 seconds of blocking and nearly 4 minutes to clear backlog for S&R contrasted with Koala sustaining input under uniform access; DRRS throughput oscillated.The test used 20M uniformly accessed keys and backlog-free reconfiguration; Koala's per-batch latency rose slightly, while DRRS's eager transfer contended with processing.Source: Figure 22 · p. 21, §D.5 · p. 21