“

Series: S8 Technology Deep Dives · Article 7 | Level: Advanced | Reading Time: 20 min

Flink State Management: RocksDB, Checkpoints, and Large-State Tuning

#TL;DR

Flink's state management is the foundation for building stateful stream processing applications; coomia-dip's Data Layer uses Flink state to maintain real-time materialized views, window aggregations, and CDC deduplication
This article deeply analyzes RocksDB State Backend internals, Checkpoint/Savepoint mechanisms, incremental Checkpoint optimizations, and tuning strategies for large-state scenarios
Covers state TTL, Timer management, State Processor API for offline analysis, and common production issues with solutions

#Table of Contents

Flink State Model Overview
State Backend Deep Comparison
RocksDB State Backend Internals
Checkpoint Mechanism Deep Dive
Incremental Checkpoints and Changelog State Backend
Savepoints and Version Compatibility
State TTL and Automatic Cleanup
Timer State Management
State Processor API for Offline Analysis
coomia-dip Large-State Tuning Practices
Key Takeaways

#1. Flink State Model Overview

#1.1 Why State Matters

In coomia-dip's data pipelines, many core operations are stateful:

CDC Deduplication: Track the latest version number for each record, discarding out-of-order older versions
Window Aggregation: Maintain intermediate aggregation results within time windows
Materialized Views: Preserve the latest computed values of derived properties
Join Operations: Buffer data from both streams for correlation

Java

// coomia-dip CDC deduplication example: using ValueState to track latest version
public class CdcDeduplicator extends KeyedProcessFunction<String, CdcEvent, CdcEvent> {

    private ValueState<Long> latestVersion;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor =
            new ValueStateDescriptor<>("latest-version", Long.class);
        latestVersion = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(CdcEvent event, Context ctx, Collector<CdcEvent> out)
            throws Exception {
        Long currentVersion = latestVersion.value();
        if (currentVersion == null || event.getVersion() > currentVersion) {
            latestVersion.update(event.getVersion());
            out.collect(event);
        }
        // Older version events are silently discarded
    }
}

#1.2 Keyed State vs Operator State

Flink provides two categories of state primitives:

Feature	Keyed State	Operator State
Scope	Per-key	Per-operator-instance
Parallelism changes	Automatic re-partition	Manual redistribution
Supported types	ValueState / ListState / MapState / ReducingState / AggregatingState	ListState / UnionListState / BroadcastState
Typical use	Dedup, windows, joins	Kafka offset tracking, source state
coomia-dip usage	CDC processing, materialized views	Pipeline connector state

#1.3 State Lifecycle

Code

Create → Update → Checkpoint Persist → Recover → TTL Expiry Cleanup
  │                                       │
  └──── Normal Processing Loop ───────────┘

In coomia-dip, a typical materialized view state lifecycle:

Create: Initialize when first receiving a change event for an ObjectType
Update: Update materialized view intermediate state on each CDC event
Persist: Checkpoint periodically snapshots state to distributed storage (MinIO/S3)
Recover: Restore from the latest Checkpoint after failure restart
Cleanup: Clear when ObjectType is deleted or TTL expires

#2. State Backend Deep Comparison

#2.1 Three State Backends

Flink 1.18+ provides three State Backends:

HashMapStateBackend (formerly MemoryStateBackend / FsStateBackend)

YAML

# flink-conf.yaml
state.backend: hashmap
state.checkpoints.dir: s3://coomia-dip-checkpoints/flink/

Storage: JVM heap memory
Serialization: Only during checkpoint
Pros: Fastest access (direct object references)
Cons: Limited by JVM heap size, GC pressure
Use case: Small state (< few GB), ultra-low latency requirements

EmbeddedRocksDBStateBackend

YAML

state.backend: rocksdb
state.backend.rocksdb.localdir: /data/flink/rocksdb
state.checkpoints.dir: s3://coomia-dip-checkpoints/flink/

Storage: Local disk (RocksDB)
Serialization: On every read/write
Pros: State size limited only by disk, supports incremental checkpoints
Cons: Serialization overhead on access
Use case: Large state (TB-scale), coomia-dip production choice

Changelog State Backend (Flink 1.16+)

YAML

state.backend: rocksdb
state.backend.changelog.enabled: true
state.backend.changelog.storage: filesystem
dstl.dfs.base-path: s3://coomia-dip-checkpoints/changelog/

Adds WAL logging on top of RocksDB
Checkpoint only needs to flush WAL increments, dramatically reducing checkpoint time
Suitable for checkpoint-interval-sensitive scenarios

#2.2 Performance Benchmark Comparison

In coomia-dip's CDC pipeline benchmark (1 million keys, ValueState):

Metric	HashMap	RocksDB	RocksDB + Changelog
Read latency P99	0.01 ms	0.15 ms	0.15 ms
Write latency P99	0.01 ms	0.25 ms	0.28 ms
Max state size	~4 GB	~2 TB	~2 TB
Checkpoint time (10GB)	45 s	12 s (incremental)	3 s
Memory footprint	High	Low	Low

#3. RocksDB State Backend Internals

#3.1 LSM-Tree Architecture

RocksDB uses a Log-Structured Merge Tree (LSM-Tree) as its storage engine:

Code

Write path:
  Key-Value → MemTable (memory) → Immutable MemTable → Flush → SST Level-0
                                                                ↓ Compaction
                                                           SST Level-1
                                                                ↓ Compaction
                                                           SST Level-2
                                                                ...

Read path:
  Search order: MemTable → Immutable MemTable → Block Cache → Level-0 SST → Level-1 SST → ...

#3.2 Column Families and State Isolation

Flink creates a RocksDB Column Family for each state descriptor:

Java

// Internal mapping
// ValueState<String> latestVersion  →  CF: "latest-version"
// MapState<String, Object> cache    →  CF: "cache"
// ListState<Event> buffer           →  CF: "buffer"

// Each CF has independent MemTable, SST files, and compaction policies

#3.3 Serialization Format

Key serialization format:

Code

┌─────────────────┬──────────────┬───────────────────┐
│ Key-Group (2B)  │ User Key     │ Namespace (opt)   │
└─────────────────┴──────────────┴───────────────────┘

Key-Group: Used for state redistribution during parallelism changes
User Key: Application-level key (e.g., objectId)
Namespace: Distinguishes different windows in windowed scenarios

#3.4 coomia-dip RocksDB Tuning Configuration

Java

@Configuration
public class FlinkRocksDBConfig {

    public static RocksDBOptionsFactory createOptionsFactory() {
        return new RocksDBOptionsFactory() {
            @Override
            public DBOptions createDBOptions(DBOptions currentOptions,
                                              Collection<AutoCloseable> handlesToClose) {
                return currentOptions
                    .setMaxBackgroundJobs(4)           // Background flush + compaction threads
                    .setMaxOpenFiles(-1)                // Unlimited open files
                    .setDbWriteBufferSize(256 * 1024 * 1024); // Global write buffer 256MB
            }

            @Override
            public ColumnFamilyOptions createColumnOptions(
                    ColumnFamilyOptions currentOptions,
                    Collection<AutoCloseable> handlesToClose) {
                BlockBasedTableConfig tableConfig = new BlockBasedTableConfig()
                    .setBlockSize(32 * 1024)            // 32KB block
                    .setBlockCacheSize(128 * 1024 * 1024) // 128MB block cache
                    .setFilterPolicy(new BloomFilter(10, false)); // 10-bit Bloom filter

                return currentOptions
                    .setTableFormatConfig(tableConfig)
                    .setWriteBufferSize(64 * 1024 * 1024)  // 64MB MemTable per CF
                    .setMaxWriteBufferNumber(3)              // Max 3 MemTables
                    .setMinWriteBufferNumberToMerge(2)       // Trigger flush at 2
                    .setCompactionStyle(CompactionStyle.LEVEL)
                    .setTargetFileSizeBase(64 * 1024 * 1024); // Level-1 SST 64MB
            }
        };
    }
}

Tuning parameters:

Parameter	Default	coomia-dip Recommendation	Rationale
`write_buffer_size`	64MB	64-128MB	CDC scenario is write-intensive
`max_write_buffer_number`	2	3	Avoid write stalls
`block_cache_size`	8MB	128-256MB	Improve read hit rate
`bloom_filter_bits`	None	10	Reduce unnecessary disk reads
`max_background_jobs`	2	4	Speed up compaction

#4. Checkpoint Mechanism Deep Dive

#4.1 Distributed Snapshot Algorithm

Flink uses a variant of the Chandy-Lamport algorithm for distributed snapshots:

Code

JobManager                    TaskManager-1             TaskManager-2
    │                              │                         │
    │──── Trigger Checkpoint ─────>│                         │
    │                              │                         │
    │                         Inject Barrier                 │
    │                              │                         │
    │                         ┌────┴────┐                    │
    │                         │ State   │                    │
    │                         │ Snapshot│                    │
    │                         └────┬────┘                    │
    │                              │── Barrier ─────────────>│
    │                              │                    ┌────┴────┐
    │                              │                    │ State   │
    │                              │                    │ Snapshot│
    │                              │                    └────┬────┘
    │<──── Ack ────────────────────│                         │
    │<──── Ack ─────────────────────────────────────────────│
    │                                                        │
    │  Checkpoint Complete                                   │

#4.2 Aligned vs Unaligned Checkpoints

Aligned Checkpoint (Exactly-Once):

Java

// Operator blocks a channel after receiving its Barrier, waits for all other Barriers
// State snapshot triggers only after all Barriers are aligned
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

Unaligned Checkpoint (Flink 1.11+):

Java

// Does not wait for Barrier alignment; snapshots state + in-flight buffered data immediately
env.getCheckpointConfig().enableUnalignedCheckpoints();
// Suitable for heavy back-pressure scenarios

coomia-dip strategy selection:

Pipeline Type	Checkpoint Mode	Rationale
CDC to Iceberg	Aligned (Exactly-Once)	Data correctness first
CDC to Doris	Unaligned	Doris supports idempotent writes; frequent back-pressure
Materialized View	Aligned	State consistency is critical

#4.3 Checkpoint Configuration Best Practices

Java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Basic configuration
env.enableCheckpointing(60_000);  // Every 60 seconds
env.getCheckpointConfig().setCheckpointTimeout(300_000);  // 5-minute timeout
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);  // Min 30s gap
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  // Max 1 concurrent
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);  // Tolerate 3 failures

// Externalized Checkpoint (retain on cancellation)
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// RocksDB incremental checkpoint
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));  // true = incremental

// Checkpoint storage
env.getCheckpointConfig().setCheckpointStorage("s3://coomia-dip-checkpoints/flink/");

#4.4 Checkpoint Monitoring Metrics

coomia-dip monitors key Checkpoint metrics via Prometheus + Grafana:

YAML

# Prometheus alerting rules
groups:
  - name: flink_checkpoint_alerts
    rules:
      - alert: CheckpointDurationHigh
        expr: flink_jobmanager_job_lastCheckpointDuration > 120000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Checkpoint duration exceeds 2 minutes"

      - alert: CheckpointFailed
        expr: increase(flink_jobmanager_job_numberOfFailedCheckpoints[10m]) > 0
        labels:
          severity: critical
        annotations:
          summary: "Checkpoint failure detected"

      - alert: CheckpointSizeSurge
        expr: |
          flink_jobmanager_job_lastCheckpointSize /
          flink_jobmanager_job_lastCheckpointSize offset 1h > 2
        labels:
          severity: warning
        annotations:
          summary: "Checkpoint size surge (possible state leak)"

#5. Incremental Checkpoints and Changelog State Backend

#5.1 Full vs Incremental Checkpoint

Full Checkpoint: Snapshots entire state each time

Code

Checkpoint-1: [Full 10GB]    → Write 10GB
Checkpoint-2: [Full 10.1GB]  → Write 10.1GB  (only 100MB changed)
Checkpoint-3: [Full 10.2GB]  → Write 10.2GB

Incremental Checkpoint: Snapshots only the delta from the last checkpoint

Code

Checkpoint-1: [Base 10GB]       → Write 10GB
Checkpoint-2: [Delta 100MB]     → Write 100MB  ✓ 99% savings
Checkpoint-3: [Delta 120MB]     → Write 120MB

#5.2 Incremental Checkpoint Implementation

RocksDB incremental checkpoint leverages SST file immutability:

Code

Checkpoint N:   SST-1, SST-2, SST-3  (already uploaded)
                                      ↓ Compaction + new writes
Checkpoint N+1: SST-1, SST-2, SST-4, SST-5  (SST-3 merged by compaction)
                ─────────────  ──────────────
                Already exists, skip   New files, upload

Key advantages:

Checkpoint time drops from O(total state) to O(delta)
Network I/O dramatically reduced
Suitable for TB-scale large state

Caveats:

SST file reference counting: old Checkpoint SSTs must not be deleted prematurely
Recovery requires rebuilding full state: Base + all incremental deltas
Recommended to use state.checkpoints.num-retained: 3 to limit retained checkpoints

#5.3 Changelog State Backend Deep Dive

The Changelog State Backend, introduced in Flink 1.16, is a revolutionary feature:

Code

Traditional incremental Checkpoint:
  State change → RocksDB → Wait for SST flush → Upload new SST

Changelog State Backend:
  State change → RocksDB (async)
               → Changelog (sync append) → Continuous upload

  On Checkpoint trigger: Only mark Changelog truncation point

Java

// Enable Changelog State Backend
Configuration config = new Configuration();
config.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, true);
config.set(
    StateChangelogOptions.CHANGE_LOG_STORAGE,
    "filesystem"  // or "memory" (for testing)
);

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment(config);

coomia-dip Changelog use case:

Materialized view pipelines need sub-second recovery time, with checkpoint interval at 10 seconds:

YAML

# Performance comparison with Changelog mode
checkpoint_interval: 10s
state_size: 50GB

# Without Changelog
checkpoint_duration_p99: 45s  # Exceeds interval, cannot complete
checkpoint_success_rate: 30%

# With Changelog
checkpoint_duration_p99: 800ms  # Only flushes Changelog delta
checkpoint_success_rate: 99.9%

#6. Savepoints and Version Compatibility

#6.1 Checkpoint vs Savepoint

Feature	Checkpoint	Savepoint
Trigger	Automatic periodic	Manual
Purpose	Fault recovery	Version upgrades, A/B testing
Format	Backend-specific (RocksDB SST)	Unified canonical format
Compatibility	Same version	Cross-version compatible
Performance	Supports incremental	Always full

#6.2 Savepoint Operations

Bash

# Trigger Savepoint
flink savepoint <jobId> s3://coomia-dip-savepoints/

# Restore from Savepoint
flink run -s s3://coomia-dip-savepoints/savepoint-xxxxx \
    -c com.onto.dataplane.pipeline.CdcPipeline \
    coomia-dip-pipeline.jar

# Allow skipping non-restorable state (when operators change)
flink run -s <savepointPath> \
    --allowNonRestoredState \
    coomia-dip-pipeline.jar

#6.3 UID Best Practices

Java

// ✅ Assign stable UIDs to every operator
DataStream<CdcEvent> deduplicated = cdcStream
    .keyBy(CdcEvent::getObjectId)
    .process(new CdcDeduplicator())
    .uid("cdc-deduplicator")          // Stable UID
    .name("CDC Deduplicator");        // Human-readable name

DataStream<MaterializedView> materialized = deduplicated
    .keyBy(CdcEvent::getObjectType)
    .process(new MaterializationProcessor())
    .uid("materialization-processor")  // Stable UID
    .name("Materialization");

// ❌ No UID: Savepoint recovery fails to match state
DataStream<CdcEvent> bad = cdcStream
    .keyBy(CdcEvent::getObjectId)
    .process(new CdcDeduplicator());  // Auto-generated UID is unstable

#7. State TTL and Automatic Cleanup

#7.1 State TTL Configuration

In coomia-dip, CDC deduplication state cannot grow indefinitely:

Java

ValueStateDescriptor<Long> descriptor =
    new ValueStateDescriptor<>("latest-version", Long.class);

StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupFullSnapshot()       // Clean during full checkpoint
    .cleanupInRocksdbCompactFilter(1000)  // Clean during RocksDB compaction
    .cleanupIncrementally(10, true)       // Incremental cleanup on each access
    .build();

descriptor.enableTimeToLive(ttlConfig);

#7.2 Three Cleanup Strategies Compared

Strategy	Trigger	Memory Overhead	Timeliness	coomia-dip Recommendation
`cleanupFullSnapshot`	Full checkpoint	None	Poor	Not recommended (uses incremental CP)
`cleanupInRocksdbCompactFilter`	Compaction	Minimal	Medium	Recommended
`cleanupIncrementally`	Each state access	Low	Good	Use in combination

#7.3 State Leak Detection and Troubleshooting

Java

// Custom MetricGroup to monitor state size
public class StateSizeMonitor extends KeyedProcessFunction<String, Event, Event> {
    private ValueState<byte[]> state;
    private transient Counter stateEntries;

    @Override
    public void open(Configuration parameters) {
        state = getRuntimeContext().getState(
            new ValueStateDescriptor<>("data", byte[].class));
        stateEntries = getRuntimeContext()
            .getMetricGroup()
            .addGroup("coomia-dip")
            .counter("state_entries");
    }

    @Override
    public void processElement(Event event, Context ctx, Collector<Event> out)
            throws Exception {
        if (state.value() == null) {
            stateEntries.inc();  // Count new keys
        }
        state.update(serialize(event));
        out.collect(event);
    }
}

#8. Timer State Management

#8.1 Event-Time Timers and Processing-Time Timers

Java

public class OntologyEventAggregator
        extends KeyedProcessFunction<String, OntologyEvent, AggregatedMetric> {

    private ValueState<AggregatedMetric> accumulator;
    private ValueState<Long> timerTimestamp;

    @Override
    public void processElement(OntologyEvent event, Context ctx,
                               Collector<AggregatedMetric> out) throws Exception {
        AggregatedMetric current = accumulator.value();
        if (current == null) {
            current = new AggregatedMetric(event.getObjectType());
            // Register Event-Time Timer: trigger at window end
            long windowEnd = event.getTimestamp() - (event.getTimestamp() % 60_000) + 60_000;
            ctx.timerService().registerEventTimeTimer(windowEnd);
            timerTimestamp.update(windowEnd);
        }
        current.merge(event);
        accumulator.update(current);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
                        Collector<AggregatedMetric> out) throws Exception {
        AggregatedMetric result = accumulator.value();
        if (result != null) {
            out.collect(result);
            accumulator.clear();
            timerTimestamp.clear();
        }
    }
}

#8.2 Timer Storage and Performance

Timers are stored in a special Column Family in the RocksDB Backend:

Event-Time Timers: Sorted by timestamp, batch-triggered when Watermark advances
Processing-Time Timers: Triggered by system clock, using a priority queue
High-volume timer scenarios: Use RocksDB Backend (timers stored on disk)

Java

// Timer backlog monitoring
env.getConfig().setAutoWatermarkInterval(200);  // 200ms Watermark update
// If Watermark lags, timers won't fire, state keeps growing

#9. State Processor API for Offline Analysis

#9.1 Reading State from a Savepoint

Java

// Use State Processor API for offline state analysis
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend();
env.setStateBackend(backend);

SavepointReader savepoint = SavepointReader.read(
    env, "s3://coomia-dip-savepoints/savepoint-abc123", backend);

// Read deduplicator operator state
DataStream<Tuple2<String, Long>> deduplicatorState = savepoint
    .readKeyedState("cdc-deduplicator", new ReaderFunction());

// Count state entries per ObjectType
deduplicatorState
    .keyBy(t -> extractObjectType(t.f0))
    .process(new CountFunction())
    .print();  // Output state distribution by ObjectType

env.execute("State Analysis");

#9.2 State Repair and Migration

Java

// Modify state and write back to new Savepoint
SavepointWriter writer = SavepointWriter.fromExistingSavepoint(
    env, "s3://coomia-dip-savepoints/savepoint-abc123", backend);

// Modify deduplication state: reset all version numbers
writer.changeKeyedState("cdc-deduplicator", new ModifierFunction());

// Add initial state for new operator
writer.withOperator(
    OperatorIdentifier.forUid("new-processor"),
    stateBootstrapTransformation
);

writer.write("s3://coomia-dip-savepoints/savepoint-repaired");
env.execute("State Repair");

#10. coomia-dip Large-State Tuning Practices

#10.1 Memory Model Configuration

YAML

# TaskManager memory configuration (16GB total)
taskmanager.memory.process.size: 16g
taskmanager.memory.flink.size: 14g
taskmanager.memory.managed.fraction: 0.4     # 5.6GB → RocksDB
taskmanager.memory.network.fraction: 0.1      # 1.4GB → Network buffers
taskmanager.memory.task.heap.size: 4g          # 4GB → User code
taskmanager.memory.task.off-heap.size: 512m    # Off-heap memory
taskmanager.memory.jvm-metaspace.size: 256m
taskmanager.memory.jvm-overhead.fraction: 0.1  # JVM overhead

# RocksDB Managed Memory allocation
state.backend.rocksdb.memory.managed: true     # Use Managed Memory
state.backend.rocksdb.memory.write-buffer-ratio: 0.5
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

#10.2 Disk I/O Optimization

YAML

# Multiple disk directories to spread I/O
state.backend.rocksdb.localdir: /ssd1/flink/rocksdb;/ssd2/flink/rocksdb

# Limit RocksDB background threads to avoid I/O saturation
state.backend.rocksdb.thread.num: 4

# Compression configuration
state.backend.rocksdb.compression.per.level: NO_COMPRESSION;NO_COMPRESSION;LZ4_COMPRESSION;LZ4_COMPRESSION;LZ4_COMPRESSION;ZSTD_COMPRESSION;ZSTD_COMPRESSION

#10.3 Serialization Optimization

Java

// Use Flink built-in types instead of Kryo (10-100x performance difference)
// ✅ Recommended: Use POJOs or Flink TypeInformation
@TypeInfo(CdcEventTypeInfoFactory.class)
public class CdcEvent {
    public String objectId;
    public String objectType;
    public long version;
    public Map<String, Object> properties;  // ⚠️ Avoid nested generics
}

// ✅ Recommended: Manual serializer registration
env.getConfig().registerTypeWithKryoSerializer(
    OntologyInstance.class,
    OntologyInstanceSerializer.class
);

// ❌ Avoid: Kryo fallback serialization
env.getConfig().disableGenericTypes();  // Enable during development, forces type checking

#10.4 Parallelism and Key Distribution

Java

// Monitor key distribution skew
public class KeyDistributionAnalyzer {

    public static void analyzeSkew(DataStream<CdcEvent> stream) {
        stream
            .keyBy(CdcEvent::getObjectType)
            .process(new KeyedProcessFunction<>() {
                private ValueState<Long> count;

                @Override
                public void open(Configuration params) {
                    count = getRuntimeContext().getState(
                        new ValueStateDescriptor<>("count", Long.class, 0L));
                }

                @Override
                public void processElement(CdcEvent e, Context ctx,
                                           Collector<String> out) throws Exception {
                    count.update(count.value() + 1);
                    if (count.value() % 100_000 == 0) {
                        out.collect(String.format(
                            "Key=%s, SubtaskIndex=%d, Count=%d",
                            e.getObjectType(),
                            getRuntimeContext().getIndexOfThisSubtask(),
                            count.value()
                        ));
                    }
                }
            })
            .print();
    }
}

Key skew solutions:

Key salting: Append random suffixes to hot keys, then aggregate twice
Local-Global aggregation: Pre-aggregate locally before global aggregation
Custom KeySelector: Combine high-cardinality fields to reduce skew

#11. Key Takeaways

Topic	Key Conclusion
State Backend selection	Use RocksDB in production; enable incremental checkpoints
Checkpoint interval	Set based on business tolerance, typically 30s-120s
Incremental checkpoint	Must enable for large-state scenarios; saves 90%+ I/O
Changelog Backend	Use when sub-second checkpoint times are needed
State TTL	CDC dedup state must have TTL to prevent leaks
Operator UIDs	Must set manually; otherwise Savepoint recovery fails
Serialization	Avoid Kryo fallback; use POJOs or custom TypeInfo
Monitoring	Checkpoint duration, size, and failure count must be alerted
Memory allocation	Managed Memory fraction at 0.4; leave room for RocksDB
Key skew	Solve with salting or Local-Global patterns

“

Next up: S8-08 dives into the Temporal workflow engine (Part 1), exploring how coomia-dip uses Temporal for durable workflows, Activity retries, and Saga compensation patterns.