Flink Kafka Rebalances and Duplicate Iceberg Sink Files
An ad platform ingests 90,000 click events/sec into a Kafka topic with 48 partitions. A Flink SQL job computes 10-minute session windows and writes impressions to an Iceberg sink with exactly-once requirements. Lag is normally about 30 seconds, but every 20-25 minutes it jumps to 6-9 minutes and then recovers. Logs show repeated rebalance messages. Each TaskManager has four slots and an 8 GB heap; checkpoints take 45-70 seconds and hold about 2.5 GB of state. The topic key is user_id, and the top 1% of users generate 22% of events. The team raised source parallelism from 48 to 96, but lag and checkpoint duration got worse. After some restarts, the sink contains duplicate impression files and dashboards double-count. The schema registry added two optional Avro fields in the last month. The business wants sub-minute freshness and no duplicates.
1. Why did increasing parallelism from 48 to 96 make the job less stable, and what would you change?
Read the worked answer
With 48 Kafka partitions, source parallelism of 96 leaves 48 source subtasks idle. The extra parallelism still creates more concurrent state, more checkpoint work, more open sink writers, and more small Iceberg files, while the hot user_id keys continue to land in a few subtasks. The rebalances are likely a symptom of TaskManagers failing or timing out during long checkpoints and GC pauses, not a sign that Kafka needs more consumers.
Return source parallelism to 48, or justify 96 only after fixing the checkpoint path. Increase TaskManager memory and use RocksDB state with incremental checkpoints. Tune checkpoint interval, timeout, and minimum pause so checkpoints complete before the next one starts. Reduce state by using bounded session windows or pre-aggregating per user before the session operator. Address skew with a two-stage aggregation or a local/global split so one user_id does not dominate a subtask. Monitor checkpoint duration, alignment time, GC time, and Kafka consumer lag together; if checkpoint duration exceeds the interval, the job will keep restarting and rebalancing regardless of parallelism.
2. How do you achieve exactly-once sink writes and eliminate duplicate impressions after restarts? Be specific about checkpoint and Iceberg commit behavior.
Read the worked answer
Flink's Iceberg sink can provide exactly-once when checkpointing is enabled and the job restores from a completed checkpoint or savepoint. Data files written since the last completed checkpoint are not committed; on restore, the job replays from the checkpointed Kafka offsets and commits once. Duplicates usually mean one of three things: the job was restarted without restoring state, two jobs wrote to the same table, or an earlier at-least-once run already loaded overlapping data.
Verify that checkpointing is enabled, the interval is shorter than the sink commit interval, and restarts use the latest completed checkpoint. Give the job a stable UID so savepoint restoration matches operators. For extra safety, assign a deterministic event_id and deduplicate in Iceberg with MERGE or equality deletes on event_id before serving dashboards. Keep a quarantine table for duplicate event_ids so you can measure replay overlap. Do not rely on file names alone: query snapshot history and row counts to distinguish duplicate files from duplicate rows.
3. The Avro schema added two optional fields. What risks does that create, and how do you roll it out without losing exactly-once semantics?
Read the worked answer
Optional fields with defaults are usually backward compatible, but the running Flink job still has to deserialize the new schema and the Iceberg table must accept the added columns. If the job restarts from a savepoint, operator state must remain compatible; changing the job graph or field order without stable UIDs can break restoration. A sink table that has not evolved will reject or drop the new fields.
Confirm the schema registry compatibility setting, add the fields with defaults, and evolve the Iceberg table explicitly before the job writes them. Test with a shadow topic and a savepoint restore in a staging job. Roll out by stopping with a savepoint, updating the job and table schema, then restoring from that savepoint. Avoid renaming or removing fields in the same change. Validate that null defaults do not violate downstream NOT NULL contracts, and compare row counts and event_id duplicates after the restore.