Broadcast State Pattern

Static configurations in streaming pipelines are often insufficient for production environments that require agility. Many real-time scenarios demand that business logic adapts to changing conditions without incurring the downtime associated with stopping, recompiling, and redeploying a job. The Broadcast State pattern in Apache Flink addresses this requirement by allowing a low-throughput control stream to dynamically update the processing logic applied to a high-throughput data stream.

This pattern fundamentally changes how state is distributed. Unlike Keyed State, which partitions data based on a hash of the key, Broadcast State replicates the state to every parallel instance of an operator. This mechanism allows you to join a non-keyed stream (the control stream) with a keyed or non-keyed data stream, making the same rules or configuration data available to all processing tasks simultaneously.

Architecture of Broadcast Streams

The architecture relies on decoupling the data ingestion from the logic definition. You typically have two streams:

1. Event Stream: The main volume of data (e.g., transactions, logs, sensor readings).
2. Broadcast Stream: A lower volume stream containing updates (e.g., fraud rules, thresholds, feature flags).

When the Broadcast Stream is connected to the Event Stream, Flink ensures that every element from the Broadcast Stream is transmitted to all parallel instances of the downstream operator. The Event Stream continues to be partitioned according to its key or simply forwarded if unkeyed.

The Control Stream replicates data to all downstream task slots, ensuring consistent logic application across partitioned data streams.

Implementing KeyedBroadcastProcessFunction

To implement this pattern, you utilize the KeyedBroadcastProcessFunction. This function merges a keyed stream with a broadcast stream and requires the implementation of two methods: processBroadcastElement and processElement.

First, you must define a MapStateDescriptor. Unlike other state types, Broadcast State must always be in the form of a Map. This structure allows the operator to look up specific configuration entries efficiently when processing events.

// Define the descriptor for the broadcast state
MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>(
    "RulesBroadcastState",
    BasicTypeInfo.STRING_TYPE_INFO,
    TypeInformation.of(new TypeHint<Rule>() {})
);

// Broadcast the rule stream
BroadcastStream<Rule> broadcastRules = ruleStream
    .broadcast(ruleStateDescriptor);

// Connect and process
dataStream
    .keyBy(data -> data.getUserId())
    .connect(broadcastRules)
    .process(new KeyedBroadcastProcessFunction<String, Transaction, Rule, Alert>() {

        @Override
        public void processBroadcastElement(Rule rule, Context ctx, Collector<Alert> out) throws Exception {
            // Update the state with the new rule
            ctx.getBroadcastState(ruleStateDescriptor).put(rule.name, rule);
        }

        @Override
        public void processElement(Transaction txn, ReadOnlyContext ctx, Collector<Alert> out) throws Exception {
            // Read the rules and apply them to the transaction
            ReadOnlyBroadcastState<String, Rule> state = ctx.getBroadcastState(ruleStateDescriptor);
            for (Map.Entry<String, Rule> entry : state.immutableEntries()) {
                if (entry.getValue().matches(txn)) {
                    out.collect(new Alert(txn, entry.getKey()));
                }
            }
        }
    });

A critical distinction exists between the access levels in these two methods. In processBroadcastElement, you have read-write access to the broadcast state using ctx.getBroadcastState. This is where you update the internal map based on new control messages.

In processElement, the context provides only a ReadOnlyBroadcastState. This restriction is enforced to maintain determinism. Since the data stream is processed in parallel across different nodes, allowing the data stream to modify the broadcast state would lead to inconsistencies where different tasks hold different versions of the "global" configuration. Only the broadcast stream, which is guaranteed to reach all nodes, is permitted to mutate this state.

Memory Management and Scalability

While Broadcast State is powerful, it incurs a specific memory footprint profile that differs from Keyed State. With Keyed State, the total state size is roughly proportional to the number of unique keys in the data. If you scale out the cluster, the state is redistributed, and the memory pressure per node decreases.

With Broadcast State, every task maintains a full copy of the broadcast data. If your configuration data grows to 1GB and you have a parallelism of 100, your cluster effectively stores 100GB of redundant data in memory. Therefore, this pattern is optimized for low-throughput, small-sized datasets (configuration, finite rule sets, exchange rates) rather than large datasets.

The following chart illustrates the memory consumption behavior as parallelism increases.

As parallelism scales, Broadcast State (red) increases linear memory usage across the cluster, whereas partitioned Keyed State (blue) remains constant in total size.

Checkpointing and Rescaling Behavior

Flink manages Broadcast State during checkpointing by ensuring that the state is part of the operator snapshot. However, because the state is identical across all parallel instances, Flink optimizes the checkpoint storage. It does not necessarily need to store $P$ copies in the distributed file system, but logical restoration must guarantee that every task possesses the state upon recovery.

When a job is rescaled (e.g., parallelism is increased from 2 to 4), Flink ensures that the new subtasks are initialized with the current Broadcast State. This is achieved by copying the state from existing checkpoints to the new task instances. This behavior guarantees that new processing slots immediately possess the complete set of rules or configurations required to process incoming data correctly without waiting for a re-transmission of the control stream.

Asynchronous Processing

It is important to note that while Flink guarantees elements within a single stream are processed in order, the relative order between the Broadcast Stream and the Data Stream is not strictly deterministic across different tasks unless synchronized by watermarks, which Broadcast State does not natively align by default.

In high-velocity environments, a race condition may occur where a transaction arrives at the exact millisecond a rule update is being broadcast. One task might process the transaction with the old rule, while another task might process a similar transaction with the new rule, depending on network arrival times. If strict ordering between control updates and data events is required, you must rely on event time and potentially buffer data in KeyedProcessFunction until the watermark passes the timestamp of the configuration change, although this adds significant latency and complexity.

For most use cases, such as dynamic filtering, A/B testing routing, or fraud detection, eventual consistency of the rule application is acceptable, provided that the rule propagation happens with sub-second latency, which standard Flink broadcasting achieves.