State Management

Stateless operations (no memory needed):
  map:    event → transform(event)
  filter: event → keep/discard
  flatMap: event → [event1, event2, ...]

Stateful operations (need memory):
  count:     need to remember running count
  sum:       need to remember running total
  average:   need count + sum
  distinct:  need set of seen values
  join:      need to buffer one stream while waiting for matches
  pattern:   need to remember sequence of events
  session:   need to track activity per user
  dedup:     need to remember processed IDs

State in Flink:
  ✅ Local to each operator instance (fast access)
  ✅ Fault-tolerant (included in checkpoints)
  ✅ Scalable (distributed across TaskManagers)
  ✅ Queryable (can be read externally)
  ✅ Supports TTL (automatic expiry)

public class FraudDetector extends KeyedProcessFunction<String, Transaction, Alert> {

    // State: last transaction amount for this user
    private ValueState<Double> lastAmountState;

    // State: count of transactions in current window
    private ValueState<Integer> txCountState;

    // State: set of merchants seen by this user
    private MapState<String, Boolean> seenMerchantsState;

    @Override
    public void open(Configuration params) {
        // Declare state with descriptors (name + type)
        lastAmountState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("last-amount", Double.class));

        txCountState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("tx-count", Integer.class));

        seenMerchantsState = getRuntimeContext().getMapState(
            new MapStateDescriptor<>("seen-merchants",
                String.class, Boolean.class));
    }

    @Override
    public void processElement(Transaction tx, Context ctx,
                               Collector<Alert> out) throws Exception {
        Double lastAmount = lastAmountState.value();
        Integer count = txCountState.value();

        // Fraud pattern: small tx followed by large tx at new merchant
        if (lastAmount != null && lastAmount < 1.0
            && tx.getAmount() > 500.0
            && !seenMerchantsState.contains(tx.getMerchant())) {
            out.collect(new Alert(tx.getUserId(), "FRAUD_SUSPECTED"));
        }

        // Update state
        lastAmountState.update(tx.getAmount());
        txCountState.update((count == null ? 0 : count) + 1);
        seenMerchantsState.put(tx.getMerchant(), true);
    }
}

// Operator state: used in sources/sinks for offset tracking
public class CountingSource implements
        SourceFunction<Long>, CheckpointedFunction {

    private long count = 0;
    private volatile boolean running = true;

    // Operator state: survives failures
    private ListState<Long> checkpointedCount;

    @Override
    public void snapshotState(FunctionSnapshotContext ctx)
            throws Exception {
        // Save current count to state during checkpoint
        checkpointedCount.clear();
        checkpointedCount.add(count);
    }

    @Override
    public void initializeState(FunctionInitializationContext ctx)
            throws Exception {
        // Restore state on recovery
        checkpointedCount = ctx.getOperatorStateStore()
            .getListState(new ListStateDescriptor<>(
                "count", Long.class));

        // On recovery, restore from checkpoint
        if (ctx.isRestored()) {
            for (Long value : checkpointedCount.get()) {
                count = value;
            }
        }
    }

    @Override
    public void run(SourceContext<Long> ctx) throws Exception {
        while (running) {
            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(count++);
            }
        }
    }
}

// Pattern: Dynamic fraud rules applied to transaction stream

// 1. Define the broadcast state descriptor
MapStateDescriptor<String, FraudRule> ruleStateDesc =
    new MapStateDescriptor<>("fraud-rules",
        String.class, FraudRule.class);

// 2. Create broadcast stream from rules topic
BroadcastStream<FraudRule> ruleStream = env
    .fromSource(kafkaRulesSource, WatermarkStrategy.noWatermarks(),
        "Rules Source")
    .broadcast(ruleStateDesc);

// 3. Connect data stream with broadcast stream
DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getUserId)
    .connect(ruleStream)
    .process(new BroadcastProcessFunction<Transaction, FraudRule, Alert>() {

        @Override
        public void processElement(Transaction tx, ReadOnlyContext ctx,
                Collector<Alert> out) throws Exception {
            // Read-only access to broadcast state
            ReadOnlyBroadcastState<String, FraudRule> rules =
                ctx.getBroadcastState(ruleStateDesc);

            // Apply all active rules to this transaction
            for (Map.Entry<String, FraudRule> entry : rules.immutableEntries()) {
                if (entry.getValue().matches(tx)) {
                    out.collect(new Alert(tx.getUserId(),
                        entry.getKey()));
                }
            }
        }

        @Override
        public void processBroadcastElement(FraudRule rule,
                Context ctx, Collector<Alert> out) throws Exception {
            // Read-write access to broadcast state
            BroadcastState<String, FraudRule> state =
                ctx.getBroadcastState(ruleStateDesc);
            state.put(rule.getId(), rule);
        }
    });

// Option 1: HashMapStateBackend (small state, fast access)
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(
    "s3://checkpoints/job-1");

// Option 2: EmbeddedRocksDBStateBackend (large state, production)
EmbeddedRocksDBStateBackend rocksDB =
    new EmbeddedRocksDBStateBackend(true); // incremental checkpoints
env.setStateBackend(rocksDB);
env.getCheckpointConfig().setCheckpointStorage(
    "s3://checkpoints/job-1");

// RocksDB tuning via flink-conf.yaml:
// state.backend.rocksdb.memory.managed: true
// state.backend.rocksdb.block.cache-size: 256mb
// state.backend.rocksdb.writebuffer.size: 128mb
// state.backend.rocksdb.writebuffer.count: 4

// Configure TTL on a state descriptor
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(24))  // Expire after 24 hours
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(
        StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupFullSnapshot()       // Clean during full checkpoints
    .cleanupInRocksdbCompaction() // Clean during RocksDB compaction
    .build();

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

// UpdateType options:
//   OnCreateAndWrite — reset TTL on create and every update
//   OnReadAndWrite   — reset TTL on any access (read or write)

// StateVisibility options:
//   NeverReturnExpired — expired state returns null (safe default)
//   ReturnExpiredIfNotCleanedUp — may return stale data (faster)

// Cleanup strategies:
//   cleanupFullSnapshot()        — remove during checkpoint (batch)
//   cleanupIncrementally(10, true) — check N entries per access
//   cleanupInRocksdbCompaction() — remove during RocksDB compaction

State Schema Evolution Support:

POJO serializer (recommended for evolvable state):
  ✅ Add new fields (initialized to default value)
  ✅ Remove fields (silently dropped)
  ❌ Cannot change field types
  ❌ Cannot rename fields
  Rules: must have no-arg constructor, public fields or getters/setters

Avro serializer:
  ✅ Full Avro schema evolution rules
  ✅ Add fields with defaults
  ✅ Remove fields
  ✅ Promote types (int → long)
  Best for complex, frequently-evolving state schemas

Kryo serializer (default fallback):
  ❌ No schema evolution support
  ❌ Any change requires state migration
  ⚠️ Avoid for production state — use POJOs or Avro

TypeInformation (Flink-native types):
  ✅ Primitives, Tuples, Row types
  ❌ Limited evolution support
  Best for simple state (counters, timestamps)

Migration strategy when schema evolution isn't supported:
  1. Take savepoint with old job
  2. Write state migration tool (read old format, write new)
  3. Start new job from migrated savepoint