Definition
- Broadcast stream – A stream marked with
broadcast(descriptor)that delivers every record to every parallel instance of the downstream operator. - BroadcastState – A map-based operator state replicated identically on every instance, updated only in
processBroadcastElement. - Read-only on main side – The main stream can only read the broadcast state in
processElement; it cannot modify it, preventing divergence. - Dynamic rules – The pattern enables live rule updates without job restart: push a new rule onto the control topic and every instance applies it immediately.
- Small dataset – The broadcast dataset must be small because it is fully replicated in memory on every parallel instance.
The National Bank of India has many fraud-checking clerks working in parallel. The risk committee occasionally publishes an updated rulebook. The broadcast state pattern is like photocopying that rulebook and placing an identical copy on every clerk's desk the moment it changes. The clerks never edit the shared rulebook while processing transactions; they only read it. A dedicated rulebook channel updates every desk's copy whenever the committee changes the rules.
Understanding the Concepts
What is the broadcast state pattern?
The broadcast state pattern distributes a small, slowly-changing dataset to every parallel instance of an operator so they all hold an identical copy and apply it consistently. It is the standard Flink answer to "how do I change business rules in a running stream job without restarting it."
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
// Step 1: Describe the broadcast state
MapStateDescriptor<String, FraudRule> ruleDescriptor =
new MapStateDescriptor<>(
"fraud-rules", String.class, FraudRule.class);
// Step 2: Mark the control stream as broadcast
BroadcastStream<FraudRule> broadcastRules =
ruleStream.broadcast(ruleDescriptor);
// Step 3: Connect main stream to broadcast stream
txnStream
.keyBy(txn -> txn.getAccountId())
.connect(broadcastRules)
.process(new FraudScreeningFunction())
.uid("fraud-screening");
How does the process function work?
The process function has two methods: processBroadcastElement handles broadcast records and is the only place that can modify the state, and processElement handles main records with read-only access to the state.
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class FraudScreeningFunction
extends KeyedBroadcastProcessFunction<
String, Transaction, FraudRule, String> {
private final MapStateDescriptor<String, FraudRule> ruleDescriptor;
public FraudScreeningFunction(
MapStateDescriptor<String, FraudRule> ruleDescriptor) {
this.ruleDescriptor = ruleDescriptor;
}
@Override
public void processBroadcastElement(FraudRule rule,
Context ctx, Collector<String> out) throws Exception {
// WRITE: update broadcast state (allowed here)
ctx.getBroadcastState(ruleDescriptor)
.put(rule.getRuleId(), rule);
// Every parallel instance gets this update
}
@Override
public void processElement(Transaction txn,
ReadOnlyContext ctx, Collector<String> out)
throws Exception {
// READ-ONLY: check transaction against rules
for (FraudRule rule :
ctx.getBroadcastState(ruleDescriptor).immutableEntries()) {
if (rule.getValue().matches(txn)) {
out.collect("ALERT: " + txn.getAccountId()
+ " violated rule " + rule.getKey());
}
}
}
}
Broadcast rule update flow:
Rule stream: {"ruleId":"R1","maxAmount":500000}
│
▼ broadcast to ALL instances
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Instance 0 │ │ Instance 1 │ │ Instance 2 │
│ rules: {R1} │ │ rules: {R1} │ │ rules: {R1} │
│ (identical) │ │ (identical) │ │ (identical) │
└──────────────┘ └──────────────┘ └──────────────┘
▲ ▲ ▲
Main transactions flow keyed by accountId
Why is the main side read-only?
The read-only restriction on the main side is deliberate: if each instance could mutate the state based on its own partition's data, the copies would diverge and the pattern would break. Only the broadcast side writes, ensuring consistency.
// processBroadcastElement: can WRITE broadcast state
ctx.getBroadcastState(ruleDescriptor).put(rule.getRuleId(), rule);
// processElement: READ-ONLY access
ctx.getBroadcastState(ruleDescriptor).immutableEntries();
// ctx.getBroadcastState(ruleDescriptor).put(...) → NOT ALLOWED
What are the constraints?
The broadcast dataset must be small because it is fully replicated in memory on every instance. There is no strict ordering guarantee between broadcast updates and main records across instances.
// GOOD: small rule set (tens to hundreds of rules)
MapStateDescriptor<String, FraudRule> ruleDesc =
new MapStateDescriptor<>("rules", String.class, FraudRule.class);
// ~100 rules × 1 KB each = 100 KB per instance
// With 100 instances = 10 MB total (fine)
// BAD: large reference table
MapStateDescriptor<String, CustomerProfile> profileDesc =
new MapStateDescriptor<>("profiles",
String.class, CustomerProfile.class);
// 10 million profiles × 1 KB = 10 GB per instance!
// With 100 instances = 1 TB total (NOT suitable for broadcast)
// Use a lookup join or async I/O instead
How do you use it for dynamic rule updates?
The key benefit is live rule updates without restarting the job. The risk team pushes a new rule onto the Kafka control topic and every instance picks it up immediately.
// Control topic: nbi-fraud-rules (Kafka)
// Main topic: nbi-transactions (Kafka)
DataStream<FraudRule> ruleStream = env
.fromSource(KafkaSource.<FraudRule>builder()
.setBootstrapServers("broker1:9092")
.setTopics("nbi-fraud-rules")
.setDeserializer(new FraudRuleDeserializer())
.build(),
WatermarkStrategy.noWatermarks(),
"fraud-rules-source");
// New rule published to Kafka:
// {"ruleId":"R99","type":"BLOCK_MERCHANT","merchantId":"7995"}
// → Every instance gets it → Next transaction to merchant 7995 is blocked
How does it interact with checkpointing?
Broadcast state is checkpointed like any other operator state. On restore, every instance gets the full broadcast state, which matches its replicated nature.
// Broadcast state is snapshotted in every checkpoint
// On restore: every instance gets the complete rule set
// No inconsistency between instances after recovery
What are common mistakes?
The most common mistake is broadcasting a large dataset, which replicates it to every instance and consumes enormous memory. Another is trying to modify broadcast state from the main processing side.
// WRONG -- broadcasting a large reference table
BroadcastStream<CustomerProfile> broadcastProfiles =
profileStream.broadcast(profileDescriptor);
// 10 million profiles replicated to every instance → OOM!
// CORRECT -- use async I/O or lookup join for large data
txnStream.keyBy(txn -> txn.getAccountId())
.process(new AsyncLookupFunction()); // query DB per key
What should a senior engineer know?
Design broadcast rules to be safe to apply slightly before or after a given main record, since there is no strict ordering guarantee across instances. If strict coordination is needed, handle it explicitly in the process function. The broadcast state pattern is the standard answer for dynamic configuration, feature flags, and fraud rules in streaming systems.
Visual Overview
Broadcast State Pattern
┌───────────────────┐
│ Control Stream │ (Kafka: nbi-fraud-rules)
│ {R1: max 500K} │
│ {R2: block 7995} │
└────────┬──────────┘
│ broadcast(ruleDescriptor)
▼
┌────────────────────────────────────────────┐
│ BroadcastStream │
│ Delivered to EVERY parallel instance │
└────────┬──────────┬──────────┬─────────────┘
│ │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│Instance 0│ │Instance 1│ │Instance 2│
│Rules: │ │Rules: │ │Rules: │
│{R1,R2} │ │{R1,R2} │ │{R1,R2} │
│(identical)│ │(identical)│ │(identical)│
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
Main stream keyed by accountId
processElement: READ-ONLY access to rules
processBroadcastElement: WRITE access to rules
Key Takeaway
The broadcast state pattern in Flink distributes a small, slowly-changing dataset to every parallel instance of an operator so they all hold an identical copy and apply it consistently. You describe the shared state with a MapStateDescriptor, mark the control stream as broadcast, connect it to the main stream, and process both in a KeyedBroadcastProcessFunction.
The crucial detail is the split of responsibilities: processBroadcastElement handles control records and is the only place that can modify the broadcast state, while processElement handles main records with read-only access. The read-only restriction prevents instances from diverging. The broadcast dataset must be small because it is fully replicated in memory, and there is no strict ordering guarantee between broadcast updates and main records.
The headline benefit is dynamic behavior without restarting the job. A team pushes a new rule onto the control topic and every instance picks it up live. This is the standard answer for changing business logic in a running stream job.
At the National Bank of India the fraud-screening job used broadcast state for rules like maximum transaction amounts and blocked merchant categories. When the risk team needed to react to a new fraud pattern, they published one rule to the Kafka control topic and every instance applied it immediately, with no restart and no inconsistency between partitions.
Remember one sentence: Broadcast state replicates a small rule set to every parallel instance via a read-only/write-only split, enabling live rule updates without job restarts.
---