State is the most expensive part of streaming; plan its schema like a database schema

A stateless operator restarts for free; a stateful one drags along checkpoints, savepoint compatibility and migrations. State is a database smeared across the job: every operator has a stable uid (otherwise the savepoint won't restore after a refactor), every state type has evolvable serialization (POJO/Avro, not the default Kryo), every key has a TTL or an explicit “lives forever” decision (otherwise state grows monotonically until OOM — Lehman, 45, within a single job).

// three decisions that cannot be made "later"
stream.keyBy(...)
    .process(new Dedupe())
    .uid("dedupe-v1");                      // 1. stable uid: renaming the class
                                            //    must not lose state

StateTtlConfig ttl = StateTtlConfig         // 2. TTL from the business horizon
    .newBuilder(Duration.ofDays(7))         //    (dedup window), not "while it fits"
    .cleanupInRocksdbCompactFilter(1000).build();

// 3. state type — POJO/Avro with explicit evolution; Kryo generic = state
//    you can't change without a full rebuild from the source (= 80)

A check before every job release: “will this code restore from the previous savepoint?” — a contract test (74) between versions of itself. If the answer is “no”, this is a state migration with its own expand — migrate — contract, not a deploy.

The cost of state is not just heap: RocksDB runs into local disk, compactions, checkpoint amplification and recovery time. On top of the three decisions come serializer snapshots, maxParallelism and key groups (set once, and it limits future resharding), and TTL cleanup semantics — lazy cleanup does not free disk at the moment of expiry.