Apache Flink Stream Processing: Event Time, Windows, State, and Exactly-Once Semantics

Apache Flink has emerged as one of the most powerful frameworks for stateful stream processing, offering true event-time semantics, expressive windowing, fault-tolerant state management, and end-to-end exactly-once delivery guarantees. This guide covers the core concepts and practical implementation patterns for production Flink deployments.

Flink Streaming Model

Flink treats every computation as a stream. Batch processing is a special case of streaming over a bounded dataset. The key APIs are:

• DataStream API: Low-level stream transformations in Java/Scala/Python
• Table API / Flink SQL: Declarative relational operations on streams
• Stateful Functions: Event-driven stateful microservices

Flink applications are represented as a logical dataflow graph: sources → transformations → sinks.

Event Time vs Processing Time vs Ingestion Time

Choosing the right time characteristic is foundational.

Processing time: Time on the machine when the event is processed. Simplest, lowest latency, but non-deterministic — replays produce different results.

Ingestion time: Time when the event enters Flink. More stable than processing time but still non-deterministic on replay.

Event time: Timestamp embedded in the event itself. Deterministic, correct for out-of-order data, but requires watermarks to handle late arrivals.

// Java DataStream API with event-time watermarks
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Event> events = env
    .fromSource(kafkaSource, WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(30))
        .withTimestampAssigner((event, ts) -> event.getTimestamp())
        .withIdleness(Duration.ofMinutes(5)),
        "Kafka Events");

Watermarks

Watermarks signal to Flink that all events with timestamps at or before the watermark value have been received. They control when windows fire and when late data is considered truly late.

public class BoundedOutOfOrdernessGenerator implements WatermarkGenerator<MyEvent> {
    private final long maxOutOfOrderness = 30_000L; // 30 seconds in ms
    private long currentMaxTimestamp;

    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
    }
}

Watermark propagation: In multi-source topologies, the minimum watermark across all parallel subtasks advances the global watermark. Idle sources can hold back the watermark indefinitely — withIdleness marks a source idle after the specified duration, preventing stalls.

Window Operations

Windows segment an infinite stream into finite chunks for aggregation.

Tumbling windows — non-overlapping, fixed size:

events.keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new CountAggregate(), new WindowResultFunction());

Sliding windows — overlapping:

events.keyBy(Event::getUserId)
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(2)))
    .sum("revenue");

Session windows — gap-based, variable size; ideal for user session analysis:

events.keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .process(new SessionAnalyzer());

Late data handling with side outputs:

OutputTag<Event> lateOutputTag = new OutputTag<Event>("late-data"){};

SingleOutputStreamOperator<AggResult> mainStream = events
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .sideOutputLateData(lateOutputTag)
    .allowedLateness(Time.minutes(1))
    .aggregate(new CountAggregate());

DataStream<Event> lateStream = mainStream.getSideOutput(lateOutputTag);
// Route late data to a correction topic or dead-letter queue

KeyedProcessFunction and Timers

KeyedProcessFunction gives full control over event processing, state, and timers — the most powerful API in Flink for complex event processing.

public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
    private ValueState<Boolean> flagState;
    private ValueState<Long> timerState;

    @Override
    public void open(Configuration params) {
        flagState  = getRuntimeContext().getState(
            new ValueStateDescriptor<>("flag", Types.BOOLEAN));
        timerState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("timer", Types.LONG));
    }

    @Override
    public void processElement(Transaction tx, Context ctx, Collector<Alert> out) throws Exception {
        Boolean lastWasSmall = flagState.value();
        if (lastWasSmall != null && lastWasSmall && tx.getAmount() > 500) {
            out.collect(new Alert(tx.getAccountId()));
            cleanUp(ctx);
        }
        if (tx.getAmount() < 1.00) {
            flagState.update(true);
            long timer = ctx.timerService().currentProcessingTime() + 60_000L;
            ctx.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }
    }

    @Override
    public void onTimer(long ts, OnTimerContext ctx, Collector<Alert> out) throws Exception {
        timerState.clear();
        flagState.clear();
    }

    private void cleanUp(Context ctx) throws Exception {
        Long timer = timerState.value();
        if (timer != null) ctx.timerService().deleteProcessingTimeTimer(timer);
        timerState.clear();
        flagState.clear();
    }
}

State Management

Flink offers multiple state backends:

• HashMapStateBackend: In-memory, fast, limited by JVM heap — suitable for small state
• EmbeddedRocksDBStateBackend: Disk-backed via RocksDB, supports TB-scale state, roughly 10x slower for point lookups but enables incremental checkpoints

env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); // true = incremental checkpoints
env.getCheckpointConfig().setCheckpointStorage("s3://flink-checkpoints/my-job/");

State types:

// ValueState — single value per key
ValueState<UserProfile> profileState = getRuntimeContext()
    .getState(new ValueStateDescriptor<>("profile", UserProfile.class));

// ListState — append-only list per key
ListState<Event> buffer = getRuntimeContext()
    .getListState(new ListStateDescriptor<>("buffer", Event.class));

// MapState — key-value map per key; efficient for sparse lookups
MapState<String, Long> countMap = getRuntimeContext()
    .getMapState(new MapStateDescriptor<>("counts", String.class, Long.class));

State TTL for automatic stale-state cleanup:

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupIncrementally(1000, true)
    .build();

ValueStateDescriptor<UserSession> desc = new ValueStateDescriptor<>("session", UserSession.class);
desc.enableTimeToLive(ttlConfig);

Checkpointing and Fault Tolerance

Flink fault tolerance is based on the Chandy-Lamport distributed snapshot algorithm. Checkpoints capture a consistent snapshot of all operator state, enabling recovery from failures by replaying source data from saved offsets.

env.enableCheckpointing(60_000); // checkpoint every 60 seconds
CheckpointConfig cfg = env.getCheckpointConfig();
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
cfg.setMinPauseBetweenCheckpoints(30_000);
cfg.setCheckpointTimeout(120_000);
cfg.setMaxConcurrentCheckpoints(1);
cfg.setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
cfg.enableUnalignedCheckpoints(); // reduces checkpoint time under backpressure

End-to-End Exactly-Once with Kafka

For exactly-once semantics from Kafka source to Kafka sink, Flink uses a two-phase commit protocol via Kafka transactions.

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("input-events")
    .setGroupId("flink-consumer-group")
    .setStartingOffsets(OffsetsInitializer.committed())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-events")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("flink-producer-")
    .build();

DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
stream.sinkTo(sink);
env.execute("Exactly-Once Pipeline");

Flink SQL for Stream Analytics

Flink SQL provides a high-level declarative interface, ideal for standard aggregation patterns.

CREATE TABLE user_events (
    user_id    BIGINT,
    event_type STRING,
    amount     DOUBLE,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'user-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format'    = 'json'
);

-- Tumbling window revenue aggregation
SELECT
    TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
    user_id,
    COUNT(*)     AS event_count,
    SUM(amount)  AS total_amount
FROM user_events
GROUP BY TUMBLE(event_time, INTERVAL '5' MINUTE), user_id;

-- Temporal join: enrich events with versioned dimension data
SELECT e.user_id, e.amount, u.tier
FROM user_events AS e
LEFT JOIN user_tiers FOR SYSTEM_TIME AS OF e.event_time AS u
ON e.user_id = u.user_id;

Async I/O for External Lookups

Blocking synchronous I/O for external enrichment is a common bottleneck. Async I/O issues multiple requests concurrently, keeping the pipeline throughput high.

DataStream<EnrichedEvent> enriched = AsyncDataStream.unorderedWait(
    events,
    new AsyncDatabaseEnricher(),
    1000,
    TimeUnit.MILLISECONDS,
    100  // max concurrent async requests
);

Backpressure Handling

Flink propagates backpressure upstream through credit-based flow control. Monitor the backpressure ratio in the Flink UI — sustained HIGH backpressure indicates a bottleneck operator.

Common remedies:

1. Increase operator parallelism of the slow stage
2. Optimize the operator (reduce state I/O, batch external calls)
3. Use Async I/O for external system lookups
4. Scale the sink target if the downstream system is saturated

Savepoints vs Checkpoints

• Checkpoints: Automatic, managed by Flink; used for failure recovery
• Savepoints: Manual, operator-triggered; used for planned upgrades, rescaling, A/B testing

# Take a savepoint before upgrading job logic
flink savepoint <job-id> s3://flink-savepoints/before-upgrade/

# Restore and start updated job from savepoint
flink run -s s3://flink-savepoints/before-upgrade/ my-updated-job.jar

Conclusion

Apache Flink's combination of event-time semantics, expressive windowing, rich state management, and end-to-end exactly-once guarantees makes it uniquely suited for mission-critical streaming applications. Start with Flink SQL for standard aggregations, use the DataStream API for complex stateful logic, configure checkpointing for fault tolerance, handle late data with side outputs and allowed lateness, and monitor backpressure to maintain throughput. These foundations enable real-time pipelines that are both correct and operationally resilient.