Apache Flink developer on retainer: KeyedProcessFunction ValueState TTL 7 days — Kafka topic 30-day retention — savepoint corruption forces earliest-offset restart — 1,032,000 duplicate user engagement events accumulate over 24 days

Apache Flink 1.18.1 • AWS Managed Service for Apache Flink • Confluent Cloud Kafka • Amazon Redshift • KeyedProcessFunction • ValueState TTL • savepoint • deduplication • DAU metric inflation • retainer billing

Meridian Stream Intelligence LLC’s product analytics team noticed an unusual pattern on October 8, 2026: the daily active user (DAU) metric had been running approximately 38% above its established baseline since September 15 — a jump that coincided with no product launch, no marketing campaign, and no significant traffic acquisition event. The anomaly was first attributed to a data pipeline fix that had shipped on September 14, which the team assumed had started capturing previously uncounted engagement events. When a second analyst cross-checked the inflated DAU figures against mobile analytics SDK logs (which counted distinct device IDs independently of the Flink pipeline) and found no corresponding increase in device-level activity, the Apache Flink developer retainer was engaged. The investigation identified 1,032,000 duplicate user engagement events in the analytics.user_engagement Redshift table, all written between September 15 and October 8. The root cause was a KeyedProcessFunction that used ValueState<Boolean> with a 7-day TTL for event_id-based deduplication. On September 15, the Flink job’s savepoint on S3 had been rendered unrestorable after S3 Intelligent-Tiering automatically moved the savepoint files to the Glacier Instant Retrieval storage class to reduce storage costs. The Flink job restarted from the earliest available Kafka offset — 30 days back in the user-engagement-events topic — and re-processed 30 days of historical events. For events in the most recent 7 days (September 8–15), the ValueState TTL had not yet expired and the deduplication check correctly recognized them as already-seen events, suppressing re-emission. For events older than 7 days (August 15 – September 7 — 24 days of data), the TTL had expired and the state was cleared: the deduplication check treated all 43,000 events per day as new, unique events and emitted them to the Redshift sink as a second copy. The 24 days × 43,000 events/day = 1,032,000 duplicate engagement records inflated the DAU metric for any day in that 24-day window.

Meridian Stream Intelligence LLC ran a Flink DataStream job on AWS Managed Service for Apache Flink (Flink 1.18.1). The job consumed a Confluent Cloud Kafka topic user-engagement-events configured with a 30-day retention period. Each event carried: event_id (UUID), user_id, session_id, event_type (page_view, click, form_submit), and event_timestamp. The Flink job keyed the stream on user_id and applied a KeyedProcessFunction that deduped events by event_id to handle Kafka’s at-least-once delivery guarantees. The deduplication state used a ValueState<Boolean> with a 7-day TTL, a choice made to limit RocksDB state size given the volume of events:

// UserEngagementDeduplicationJob.java
// Apache Flink 1.18.1, AWS Managed Service for Apache Flink

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class EventDeduplicationFunction
        extends KeyedProcessFunction<String, UserEngagementEvent, UserEngagementEvent> {

    private ValueState<Boolean> seenState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Boolean> descriptor =
            new ValueStateDescriptor<>("seen-event-ids", Boolean.class);

        // TTL: 7 days — chosen to keep RocksDB state size manageable
        // PROBLEM: Kafka topic user-engagement-events has 30-day retention
        // If the job restarts from earliest offset, events up to 30 days old are replayed
        // Events older than 7 days have expired TTL: seenState.value() returns null
        // Deduplication check fails → all events older than 7 days are re-emitted
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.days(7))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
        descriptor.enableTimeToLive(ttlConfig);

        seenState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(UserEngagementEvent event,
                                Context ctx,
                                Collector<UserEngagementEvent> out) throws Exception {
        if (seenState.value() == null) {
            // Event not seen before (or TTL expired — indistinguishable)
            seenState.update(Boolean.TRUE);
            out.collect(event);
        }
        // else: duplicate — suppress
    }
}

The Flink job wrote deduplicated events to Amazon Redshift via Amazon Kinesis Data Firehose. The user-engagement-events Kafka topic was managed on Confluent Cloud with the organization’s standard retention policy of 30 days. The Flink job was configured with RocksDB state backend, with savepoints stored at s3://msi-flink-savepoints/user-engagement/. On August 14, a successful savepoint had been taken at path s3://msi-flink-savepoints/user-engagement/sp-2026-08-14/. The S3 bucket had an Intelligent-Tiering lifecycle policy applied to all objects older than 30 days: after 30 days, objects with no access were transitioned to the Glacier Instant Retrieval tier to reduce storage costs. On September 14 — exactly 31 days after the August 14 savepoint — the S3 Intelligent-Tiering policy moved all objects under sp-2026-08-14/ to Glacier Instant Retrieval.

On September 15, a routine Flink job configuration update (adding a new output metric) required a job restart. The Flink job operator triggered a stop-with-savepoint, which failed because the existing savepoint path pointed to the August 14 savepoint, which was now in Glacier Instant Retrieval and could not be read by the Flink S3 file system plugin without an explicit S3 Glacier restore operation. The operator, unfamiliar with the Glacier transition behavior, attempted to restart the Flink job from the August 14 savepoint path and received a FileNotFoundException. With the savepoint unavailable, the restart policy defaulted to consuming from the earliest available Kafka offset — 30 days back in the user-engagement-events topic, corresponding to August 15:

// September 15 restart error — savepoint in Glacier Instant Retrieval:
// org.apache.flink.runtime.checkpoint.CheckpointException:
//   Could not restore from checkpoint / savepoint.
//   Caused by: java.io.FileNotFoundException:
//   s3://msi-flink-savepoints/user-engagement/sp-2026-08-14/_metadata
//   (The specified key does not exist or is not accessible in the current storage class)

// Operator restarts job with startingOffsets = "earliest"
// Kafka consumer begins replaying from August 15 (topic retention = 30 days)

// Events replayed:
// August 15 – September 7: 24 days × 43,000 events/day = 1,032,000 events
//   → seenState.value() = null (TTL expired after 7 days)
//   → all 1,032,000 events emitted to Kinesis Firehose → Redshift
//   → DUPLICATE records in analytics.user_engagement

// August 15 – September 7 in Kafka: real events from original ingest
// August 15 – September 7 in Redshift: now appear twice each

// September 8 – September 14: 7 days × 43,000 events/day = 301,000 events
//   → seenState.value() = true (TTL not yet expired — within 7-day window)
//   → correctly suppressed ✓

// September 15+: new events, no duplicates ✓

// Net effect: 1,032,000 duplicate records written to analytics.user_engagement

The Kinesis Firehose connector wrote all events received from the Flink job to Redshift without deduplication at the sink layer. Redshift’s COPY command loaded the Firehose S3 output files into analytics.user_engagement without any MERGE or UPSERT logic — the table had no UNIQUE constraint on event_id because the Flink job was designed to guarantee uniqueness before emission, eliminating the perceived need for a database-level constraint. The result was 1,032,000 duplicate rows in Redshift that were syntactically valid records (same event_id, same user_id, same timestamp as the originals) and therefore passed all standard data quality checks that looked only at NULL counts, range validation, and referential integrity.

Why state TTL shorter than Kafka retention is a latent deduplication failure

The structural issue is that the 7-day state TTL and the 30-day Kafka retention period create two incompatible invariants. The TTL was set to limit RocksDB state size: at 43,000 events per day with a UUID event_id (16 bytes) as the state key, a 30-day TTL would require storing approximately 1.3 million keyed entries in RocksDB, consuming an estimated 350 MB of state (after RocksDB compaction), compared to approximately 80 MB at 7-day TTL. The Kafka retention was set to 30 days by the organization’s standard data pipeline policy, which treated topic retention as an independent storage concern unrelated to the Flink job’s state configuration.

The key insight is that the state TTL must be set relative to the worst-case re-processing window, not the normal operating window. During normal operation, the Flink job processes events in near-real-time; the oldest event in any given micro-batch is at most a few minutes old, well within any reasonable TTL. The TTL only matters in the failure scenario: when the job must restart from a historical Kafka offset. In that scenario, the re-processing window is bounded by the topic retention period, not by the job’s normal latency. Because a 7-day TTL is shorter than the 30-day topic retention, any restart from an offset older than 7 days — which includes any restart from earliest offset on a 30-day-retention topic — will produce duplicates for the events that fall outside the TTL window.

// Diagnosis Step 1: Confirm duplicate event_ids in Redshift user_engagement
SELECT COUNT(*) AS total_events,
       COUNT(DISTINCT event_id) AS unique_events,
       COUNT(*) - COUNT(DISTINCT event_id) AS duplicate_events
FROM analytics.user_engagement;
-- total_events:     48,032,000
-- unique_events:    47,000,000
-- duplicate_events:  1,032,000  ✓

// Diagnosis Step 2: Identify when duplicates were written
SELECT DATE_TRUNC('day', event_timestamp) AS event_date,
       COUNT(*) - COUNT(DISTINCT event_id) AS duplicate_count,
       MIN(event_timestamp) AS earliest_dup,
       MAX(event_timestamp) AS latest_dup
FROM analytics.user_engagement
GROUP BY DATE_TRUNC('day', event_timestamp)
HAVING COUNT(*) > COUNT(DISTINCT event_id)
ORDER BY event_date;
-- Results: August 15, 2026 through September 7, 2026
-- 24 calendar days × ~43,000 duplicates/day = 1,032,000 ✓
-- (September 8–14: no duplicates — within 7-day TTL window at restart time)

// Diagnosis Step 3: Correlate with Flink job restart timestamp
// AWS Managed Service for Apache Flink CloudWatch Logs:
// 2026-09-15T11:42:07Z INFO JobManager: Attempting restore from savepoint
// 2026-09-15T11:42:09Z ERROR CheckpointCoordinator: Failed to restore savepoint
//   s3://msi-flink-savepoints/user-engagement/sp-2026-08-14/_metadata
//   java.io.FileNotFoundException (Glacier Instant Retrieval — not directly accessible)
// 2026-09-15T11:42:11Z WARN KafkaConsumerThread: No savepoint available;
//   starting from earliest offset in user-engagement-events
// → First event replayed from: 2026-08-15T00:00:03Z (30-day retention boundary)

// Diagnosis Step 4: Verify TTL configuration
// KeyedProcessFunction source: EventDeduplicationFunction.java
// TTL: Time.days(7)
// Kafka retention: 30 days
// Gap: 30 - 7 = 23 days of events outside TTL window → 24 calendar days × 43,000 = 1,032,000 ✓

// Diagnosis Step 5: Confirm DAU inflation
SELECT DATE_TRUNC('day', event_timestamp) AS activity_date,
       COUNT(DISTINCT user_id) AS dau_with_duplicates
FROM analytics.user_engagement
WHERE event_timestamp BETWEEN '2026-09-15' AND '2026-10-08'
GROUP BY DATE_TRUNC('day', event_timestamp)
ORDER BY activity_date;
-- DAU values approximately 38% above pre-September-15 baseline ✓
-- (duplicate events attributed to same user_ids push COUNT(DISTINCT user_id) up
--  for users who appeared in both the original and re-emitted event sets)

Diagnosis was complete in 2.0 hours. The root cause was the combination of: (1) ValueState TTL configured at 7 days, shorter than the 30-day Kafka topic retention period, and (2) a savepoint made unrestorable by an S3 Intelligent-Tiering lifecycle policy that transitioned savepoint files to Glacier Instant Retrieval after 30 days of no access. The two conditions were operationally independent — the S3 lifecycle policy was applied to all buckets as a cost-reduction measure, not specifically to the savepoints bucket, and the Flink team did not flag the savepoints path as an exception. The TTL choice was documented as a state size optimization but was not cross-referenced against the Kafka topic retention policy in any operational runbook.

The fix: extend state TTL to match Kafka retention, add S3 Object Lock to savepoints, delete 1,032,000 duplicate rows

Part 1: Extend ValueState TTL to 32 days. The TTL was updated to 32 days — 30 days to match Kafka topic retention plus a 2-day safety margin to account for any topic retention extension or delay in TTL expiry. At 43,000 events per day, the RocksDB state size increase was estimated at approximately 280 MB (from 80 MB to 360 MB total state), acceptable for the job’s task manager memory allocation of 4 GB. Incremental RocksDB checkpointing was already enabled; no additional configuration was needed to support the larger state size.

// Fixed EventDeduplicationFunction.java — TTL extended to 32 days
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(32))  // was: Time.days(7)
    // Rule: TTL must be >= Kafka topic retention (30 days) + safety margin (2 days)
    // If Kafka retention changes, this TTL must be updated to match.
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
descriptor.enableTimeToLive(ttlConfig);

// Additionally: configure RocksDB incremental checkpoints
// (already enabled; verify checkpoint interval and storage path)
// AWS MWAA / Managed Service for Apache Flink application config:
// "CheckpointingEnabled": true,
// "CheckpointInterval": 60000,  // 60 seconds
// "MinPauseBetweenCheckpoints": 5000,
// "CheckpointingMode": "EXACTLY_ONCE"
// StateBackend: RocksDB with incremental checkpointing on S3 Standard (not Intelligent-Tiering)

Part 2: Protect savepoints from S3 lifecycle policies. The savepoints S3 bucket was updated to exclude the /user-engagement/ prefix from the Intelligent-Tiering lifecycle policy. An S3 Object Lock policy (Governance mode, 90-day retention period) was added to all savepoint files to prevent accidental deletion or storage class transition while the savepoint is potentially needed for job recovery. A separate bucket tag Flink-Savepoints: true was added to enable easy identification of savepoint buckets during future infrastructure audits.

Part 3: Delete the 1,032,000 duplicate rows from Redshift. Because Redshift does not support a direct DELETE WHERE ROW_NUMBER() > 1 syntax, the deduplication was performed using a staging table approach:

-- Redshift deduplication: keep earliest row per event_id, delete duplicates
-- Step 1: Create staging table with one row per event_id (earliest by ingest_timestamp)
CREATE TABLE analytics.user_engagement_deduped AS
SELECT *
FROM (
    SELECT *,
           ROW_NUMBER() OVER (
               PARTITION BY event_id
               ORDER BY ingest_timestamp ASC   -- keep first-ingested record
           ) AS rn
    FROM analytics.user_engagement
) t
WHERE rn = 1;
-- Rows in deduped table: 47,000,000 (= total unique event_ids) ✓

-- Step 2: Verify row counts before swap
SELECT COUNT(*) FROM analytics.user_engagement;        -- 48,032,000
SELECT COUNT(*) FROM analytics.user_engagement_deduped; -- 47,000,000
-- Difference: 1,032,000 duplicates to be removed ✓

-- Step 3: Atomic swap
BEGIN;
ALTER TABLE analytics.user_engagement RENAME TO user_engagement_bak_20261009;
ALTER TABLE analytics.user_engagement_deduped RENAME TO user_engagement;
COMMIT;

-- Step 4: Validate post-swap
SELECT COUNT(*) AS total,
       COUNT(DISTINCT event_id) AS unique_count
FROM analytics.user_engagement;
-- total: 47,000,000    unique_count: 47,000,000  ✓ (no duplicates)

-- Step 5: Re-run DAU for September 15 – October 8 to confirm correct values
SELECT DATE_TRUNC('day', event_timestamp) AS activity_date,
       COUNT(DISTINCT user_id) AS dau
FROM analytics.user_engagement
WHERE event_timestamp BETWEEN '2026-09-15' AND '2026-10-08'
GROUP BY DATE_TRUNC('day', event_timestamp)
ORDER BY activity_date;
-- DAU values now within 1.5% of pre-September-15 baseline ✓ (anomaly resolved)

-- Step 6: Drop backup table after 7-day hold
-- DROP TABLE analytics.user_engagement_bak_20261009;  -- schedule for 2026-10-16

Total engagement time: 5.5 hours (2.0h diagnosis: duplicate event_id count query on analytics.user_engagement confirming 1,032,000 duplicates; time-series analysis of duplicate distribution confirming 24-day window from August 15 to September 7; correlation with Flink job CloudWatch logs confirming September 15 savepoint restoration failure and earliest-offset restart; KeyedProcessFunction TTL audit confirming 7-day TTL vs. 30-day Kafka retention mismatch; S3 bucket lifecycle policy review confirming Intelligent-Tiering transition of savepoint files on September 14; 2.5h fix: TTL extended to 32 days; S3 lifecycle policy exclusion for savepoints prefix; S3 Object Lock governance policy applied to savepoints bucket; Flink job restart from manually restored savepoint (S3 Glacier restore request initiated, 12-hour standard retrieval, job temporarily paused during retrieval); job restart validated: no new duplicates in first 4-hour monitoring window; 1.0h data recovery and validation: Redshift deduplication via ROW_NUMBER staging table approach; atomic table swap; post-swap count validation confirming 47,000,000 total = 47,000,000 unique; DAU re-run for 24-day window confirming values within 1.5% of pre-incident baseline).

Why the Flink state TTL vs. Kafka retention gap is a recurring retainer failure pattern

The Meridian Stream Intelligence incident illustrates a structural gap between Flink state management and Kafka topic retention configuration that is almost never audited in practice. Flink state TTL configuration is treated as a Flink concern — it lives in the Flink job code, is set by the Flink developer, and is reviewed in the context of Flink state size management. Kafka topic retention is treated as a Kafka concern — it is set in Kafka topic configuration, managed by the Kafka or platform team, and reviewed in the context of storage cost. Neither team has a natural reason to cross-reference the two values, and no standard Flink metric or Kafka metric surfaces the relationship between them.

The failure mode is specifically triggered by the savepoint failure path: the combination of a corrupted or inaccessible savepoint and a long-retention Kafka topic. This combination is operationally plausible for multiple reasons beyond S3 Intelligent-Tiering: a Flink cluster upgrade that uses an incompatible savepoint format, a savepoint taken on a different Flink version, a savepoint stored on a volume that was lost during a cluster migration, or an operator who simply does not know where the most recent valid savepoint is located. In any of these scenarios, the job falls back to the earliest Kafka offset, and any state TTL shorter than the topic retention period becomes a deduplication gap.

An Apache Flink developer retainer who performs a quarterly state TTL audit — explicitly comparing each stateful operator’s TTL against the corresponding source topic’s retention period, and flagging any TTL shorter than that retention period — catches this class of issue before the first savepoint failure rather than after 1,032,000 duplicates have accumulated in a production analytics table.

How HourTab helps Apache Flink developer retainers

The Meridian Stream Intelligence engagement totaled 5.5 hours: diagnosis (2.0h — 0.25h duplicate event_id count query on analytics.user_engagement confirming 1,032,000 duplicates in 47,000,000 + 1,032,000 = 48,032,000 total rows; 0.5h time-series analysis of duplicate distribution by event_date confirming 24-day window from August 15 to September 7 matching events where ValueState TTL had expired at restart time; 0.5h Flink job CloudWatch log analysis confirming September 15 savepoint restoration failure (FileNotFoundException on sp-2026-08-14/_metadata) and earliest-offset Kafka restart; 0.25h KeyedProcessFunction source code audit confirming TTL set to Time.days(7) vs. Kafka topic retention of 30 days; 0.5h S3 bucket lifecycle policy review in AWS Console confirming Intelligent-Tiering policy applied bucket-wide including savepoints prefix, transition timestamp September 14), fix (2.5h — 0.5h TTL code change: StateTtlConfig updated to Time.days(32); Flink job package build and deployment to AWS Managed Service for Apache Flink; 0.5h S3 lifecycle policy update: added Filter to exclude /msi-flink-savepoints/ prefix from Intelligent-Tiering transitions; S3 Object Lock Governance mode policy applied; 0.5h S3 Glacier restore request for sp-2026-08-14 (standard retrieval, estimated 12h); Flink job paused pending savepoint restore; 1.0h post-restore job restart with restored savepoint; 4-hour monitoring window confirming no new duplicates written to Kinesis Firehose), and data recovery (1.0h — 0.5h Redshift deduplication: ROW_NUMBER staging table, 47,000,000-row deduped table created, atomic rename; 0.5h post-swap validation: COUNT(*) = COUNT(DISTINCT event_id) = 47,000,000 confirmed; DAU re-run for 23-day anomaly window showing values within 1.5% of pre-incident baseline).

In a HourTab retainer log, the incident session reads: “analytics.user_engagement duplicate events: Flink 1.18.1 AWS Managed Service for Apache Flink; Confluent Cloud user-engagement-events topic (30-day retention); KeyedProcessFunction ValueState TTL 7 days (mismatch: TTL < Kafka retention); savepoint sp-2026-08-14 transitioned to S3 Glacier Instant Retrieval by Intelligent-Tiering policy on September 14; September 15 restart: FileNotFoundException on savepoint → earliest-offset restart; 30 days replayed (August 15 – September 15); events > 7 days old: TTL expired → deduplication check fails → all re-emitted; events <= 7 days old: TTL valid → correctly suppressed; 24 days × 43,000 events/day = 1,032,000 duplicates; DAU inflated 38% for September 15–October 8; discovery October 8 mobile SDK cross-check; fix: TTL extended to 32 days; S3 lifecycle exclusion for savepoints; Object Lock governance; DELETE 1,032,000 duplicate rows via ROW_NUMBER staging; DAU within 1.5% of baseline confirmed; 5.5h.”

HourTab turns a time-tracker CSV into a public retainer-hours URL your client can bookmark — no login, no portal, no “how many hours do I have left?” emails. Free for one retainer. $9/mo for up to ten.

FAQ: Apache Flink developer retainers

See also: Kafka developer retainer · Elasticsearch developer retainer · Apache Airflow developer retainer