Phase 3: Stateful Stream Processing & State Backends¶

Audience: Principal Distributed Systems Architect & High-Scale Systems Engineer
Source References: flink-runtime/src/main/java/org/apache/flink/runtime/state/, flink-state-backends/

1. Keyed State vs. Operator State¶

Flink organizes all state into two fundamental categories:

┌────────────────────────────────────────────────────────────────────────┐
│ Keyed State (Available strictly after a keyBy() operation)             │
├────────────────────────────────────────────────────────────────────────┤
│ - Partitioned and sharded across Task Slots by key hash                │
│ - ValueState<T>: Single value per key (e.g., account balance)          │
│ - ListState<T>: List of elements per key (e.g., historical user events)│
│ - MapState<UK, UV>: Key-value dictionary per key                       │
│ - ReducingState<T> / AggregatingState<IN, OUT>: Pre-aggregated values  │
└────────────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────────────┐
│ Operator State (Scoped per parallel task instance)                     │
├────────────────────────────────────────────────────────────────────────┤
│ - Bound to a specific parallel worker thread                           │
│ - ListState<T>: List of records redistributed on rescaling             │
│ - UnionListState<T>: Every task gets a copy of ALL state on rescaling  │
│ - BroadcastState<K, V>: Broadcasted to ALL parallel operator tasks    │
│   (Used for dynamic runtime rules and ML model weight updates)         │
└────────────────────────────────────────────────────────────────────────┘

2. State Backends Compared: `HashMap` vs. `EmbeddedRocksDB`¶

┌────────────────────────────────────────────────────────────────────────┐
│ 1. HashMapStateBackend (In-Memory Heap Engine)                         │
├────────────────────────────────────────────────────────────────────────┤
│ - Location: JVM Young & Old Generation Heap                            │
│ - Representation: Raw Java objects (No serialization on read/write)   │
│ - Read/Write Latency: Nanoseconds (DRAM access)                        │
│ - Limitation: Hard ceiling bound by JVM Heap RAM; GC pause overhead    │
│ - Checkpointing: Heavy serialization CPU spike during snapshot         │
└────────────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────────────┐
│ 2. EmbeddedRocksDBStateBackend (Out-of-Core Off-Heap Engine)           │
├────────────────────────────────────────────────────────────────────────┤
│ - Location: Off-Heap C++ Native Memory + Local NVMe/SSD Storage        │
│ - Representation: Serialized binary byte arrays (TypeSerializer)       │
│ - Read/Write Latency: Microseconds (Cache hit) to Milliseconds (Disk)  │
│ - Capacity: Terabytes of state per TaskManager (Spills to disk)        │
│ - Checkpointing: Blazing fast asynchronous SSTable hard-link copy!     │
└────────────────────────────────────────────────────────────────────────┘

2.1 The Serialization Trade-off¶

With EmbeddedRocksDBStateBackend, every call to state.value() or state.update(v) requires: 1. Serializing Java object to byte array using Flink TypeSerializer. 2. Invoking JNI (Java Native Interface) barrier into native C++ RocksDB. 3. RocksDB searching its Memtable (SkipList) or Block Cache.

Design Rule: For ultra-low latency (\(< 1\text{ms}\)) with small state (\(< 10\text{GB}\)), use HashMapStateBackend. For high-volume state (\(> 100\text{GB}\)) or unbounded retention, use EmbeddedRocksDBStateBackend.

3. State Time-To-Live (State TTL) & RocksDB Compaction Filters¶

In unbounded streaming pipelines, state for inactive keys accumulates indefinitely, eventually causing disk or heap exhaustion.

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000) // Native C++ RocksDB background pruning!
    .build();

ValueStateDescriptor<UserSession> descriptor = 
    new ValueStateDescriptor<>("user-session", UserSession.class);
descriptor.enableTimeToLive(ttlConfig);

3.1 The 8-Byte Timestamp Prefix¶

When State TTL is enabled: - Flink prepends an 8-byte timestamp (milliseconds since epoch) to every state value byte array in RocksDB: [ 8-Byte Expiration Timestamp | Serialized Payload Bytes ] - Lazy Cleanup: When state.value() is called, Flink checks the timestamp. If expired, it returns null and frees the entry. - Active RocksDB Compaction Filter: A native C++ compaction filter inspects the 8-byte prefix during background SSTable compactions and drops expired keys directly from disk with zero JVM CPU overhead!