flink-java

Follows java-skill conventions: records for data, sealed interfaces for sum types, Result<T> for fallible operations.

Version Baseline — Flink 2.x Only

• Flink 2.x (current stable 2.2.x). Kafka connector: flink-connector-kafka 5.0.0-2.2 for Flink 2.2.x, 5.0.0-2.1 for 2.1.x; 4.0.1-2.0 is for Flink 2.0.x only — match the suffix to the Flink minor.
• Java 17 recommended, 11 minimum, 8 support removed.
• REMOVED in 2.0 — never emit: SourceFunction/SinkFunction/addSource/addSink (use Source/Sink V2 via fromSource/sinkTo), FlinkKafkaConsumer/FlinkKafkaProducer, the Scala DataStream API, env.setStateBackend (use Configuration + env.configure), per-job deployment mode.

When to Use DataStream vs SQL

Rule: Start with SQL. Drop to DataStream only when SQL can't express it.

Operator UIDs — CRITICAL

// BAD: state cannot be recovered after code changes
stream
    .keyBy(Event::key)
    .process(new StatefulProcessor())
    .print();

// GOOD: always set UID on stateful operators
stream
    .keyBy(Event::key)
    .process(new StatefulProcessor())
    .uid("stateful-processor")
    .name("Stateful Processor")
    .print();

Rule: Every stateful operator MUST have a stable, unique uid(). Without it, savepoint recovery breaks.

Records for Flink Data Types

Use records for all event, result, and error types. These four are the running data model for every example in this skill:

public record Event(String eventId, String userId, String eventType, long timestamp) {}

public record CountResult(String key, long count) {}

public record EnrichedEvent(Event event, String userName, String country) {}

public record DeadLetter(String rawPayload, String errorType, String errorMessage, Instant failedAt) {
  public static DeadLetter from(String raw, Failure f) {
    return new DeadLetter(raw, f.code(), f.message(), Instant.now());
  }
}

Flink-specific serialization facts:

• PojoSerializer recognizes records since 1.19 (FLINK-32380), serializing through the canonical constructor — public records avoid Kryo without any registration.
• Set pipeline.generic-types: false in dev configs to fail fast the moment any type would fall back to Kryo.
• State schema evolution (add/remove fields across savepoints) works only for POJO/record and Avro serialized state. Kryo-serialized state cannot evolve.

Result<T> at Flink Boundaries

Flink's operator API is exception-based (processElement throws Exception). Use Result<T> for internal logic — deserialization, validation, enrichment — then route errors to side outputs at the operator boundary.

// Parse raw JSON into Event using Result
Result<Event> parseEvent(String raw) {
  return Result.of(() -> MAPPER.readValue(raw, Event.class));
}

// Validate business rules
Result<Event> validateEvent(Event event) {
  if (event.userId() == null || event.userId().isBlank()) {
    return Result.err("VALIDATION", "missing userId");
  }
  if (event.timestamp() <= 0) {
    return Result.err("VALIDATION", "invalid timestamp");
  }
  return Result.ok(event);
}

Side Output DLQ Pattern

Route parsing and validation failures to a dead letter queue via side outputs.

private static final OutputTag<DeadLetter> DLQ = new OutputTag<>("dlq") {};

// raw: DataStream<String> from Kafka — source builder in "Kafka Integration" below
SingleOutputStreamOperator<Event> events = raw
    .process(new ProcessFunction<String, Event>() {
      @Override
      public void processElement(String value, Context ctx, Collector<Event> out) {
        switch (parseEvent(value).flatMap(e -> validateEvent(e))) {
          case Ok(var event) -> out.collect(event);
          case Err(var f) -> ctx.output(DLQ, DeadLetter.from(value, f));
        }
      }
    })
    .uid("parse-validate");

// Assign watermarks after parsing
DataStream<Event> watermarked = events
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((e, ts) -> e.timestamp())
            .withIdleness(Duration.ofMinutes(1)))
    .uid("watermarks");

// Route DLQ to separate sink
events.getSideOutput(DLQ)
    .sinkTo(dlqSink)
    .uid("dlq-sink");

Rule: Never drop data silently. Parse with Result, route failures to DLQ with full context.

Time Semantics

// Event time with bounded out-of-orderness (default choice)
WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, ts) -> event.timestamp())
    .withIdleness(Duration.ofMinutes(1));

// Watermark alignment: bounds drift between sources when one Kafka partition lags.
// FLIP-27 sources only, must be applied AT the source — not via
// assignTimestampsAndWatermarks. Idleness alone does not bound drift.
WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withWatermarkAlignment("group", Duration.ofSeconds(20), Duration.ofSeconds(1));

// Processing time — use ONLY when:
// - Latency matters more than correctness
// - Data has no meaningful timestamps
// - You never need to reprocess

Rule: Use event time unless you have a specific reason not to.

Late Data Handling

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

SingleOutputStreamOperator<CountResult> counts = stream
    .assignTimestampsAndWatermarks(watermarkStrategy)
    .keyBy(Event::userId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofMinutes(5))
    .sideOutputLateData(lateDataTag)
    .aggregate(new CountAgg(), new EmitWithKey())
    .uid("windowed-count");

// Process late data separately
counts.getSideOutput(lateDataTag)
    .sinkTo(lateDataSink)
    .uid("late-data-sink");

// Records are immutable — .sum("field") sets fields reflectively and fails on them.
// Incremental aggregate: O(1) state per key+window, no element buffering.
static final class CountAgg implements AggregateFunction<Event, Long, Long> {
  @Override public Long createAccumulator() { return 0L; }
  @Override public Long add(Event e, Long acc) { return acc + 1; }
  @Override public Long getResult(Long acc) { return acc; }
  @Override public Long merge(Long a, Long b) { return a + b; }
}

static final class EmitWithKey
    extends ProcessWindowFunction<Long, CountResult, String, TimeWindow> {
  @Override
  public void process(String key, Context ctx, Iterable<Long> aggregated,
      Collector<CountResult> out) {
    out.collect(new CountResult(key, aggregated.iterator().next()));
  }
}

Rule: Always define a late data strategy. Never silently drop data.

State Management

State TTL Configuration

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupFullSnapshot()
    .build();

ValueStateDescriptor<Long> descriptor =
    new ValueStateDescriptor<>("count", Long.class);
descriptor.enableTimeToLive(ttlConfig);

State Type Selection

// ValueState<T>: single value per key (counters, last-seen)
ValueState<Long> count;

// MapState<K,V>: key-value lookups, deduplication
MapState<String, Event> seen;

// ListState<T>: event buffering (each element serialized independently)
ListState<Event> buffer;

// ReducingState/AggregatingState: incremental computation
ReducingState<Long> sum;

Rule: Always configure TTL for user-keyed state. Use ListState over ValueState<List<T>>. Annotate per-key state growth on every stateful operator (// state: O(...) per key) — state size per key is the memory bound that kills jobs.

Comment Doctrine

The // state: O(...) per key note above is one case of a wider rule: every operator method that does real work (processElement, onTimer, add, process) gets a header comment above it — one line saying what it is for, plus its big-O for compute AND memory. The big-O lives in the header, not on a line inside the body. Tiny obvious helpers are exempt.

// Buffer one event and reset the session gap timer.
// compute: O(1) per event. memory: O(events per session) per key — ListState holds the whole session.
@Override
public void processElement(Event event, Context ctx, Collector<Session> out) throws Exception {

• Comment for the person reading and reviewing the code: the public API, anything that is not obvious, and the reason behind the code as a plain sentence (no WHY: label).
• Keep the work plan out of comments: no phases, plan steps, ticket numbers, or words like "for now", "temporary", or "later". They go stale and someone has to find and delete them. Unfinished work belongs in plan.md.
• Editing counts too: when you change an operator, keep the comments already there, fix any your change made wrong, and add the missing header, big-O, or reason. Do this only for code you add or change, not for nearby code you did not touch.

Checkpointing Configuration

env.enableCheckpointing(60_000);

var config = env.getCheckpointConfig();
// CheckpointingMode import: org.apache.flink.core.execution.CheckpointingMode
// (the streaming.api enum and setCheckpointingMode are gone in 2.x)
config.setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE);
config.setMinPauseBetweenCheckpoints(30_000);
config.setCheckpointTimeout(600_000);
config.setMaxConcurrentCheckpoints(1);
config.setTolerableCheckpointFailureNumber(3);

// CRITICAL: retain checkpoints on cancellation for recovery
// (setExternalizedCheckpointCleanup/ExternalizedCheckpointCleanup removed in 2.0)
config.setExternalizedCheckpointRetention(
    ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION
);

// RocksDB + incremental for production. env.setStateBackend was removed in
// 2.0 — Configuration + env.configure is the only way.
var conf = new Configuration();
conf.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
conf.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
env.configure(conf);

Rule: Always externalize checkpoints. Use RocksDB + incremental for production.

ProcessFunction Patterns

KeyedProcessFunction with Metrics

public final class EventCounter extends KeyedProcessFunction<String, Event, CountResult> {

  private ValueState<Long> countState;
  private transient Counter processedEvents;

  @Override
  public void open(OpenContext ctx) {
    var descriptor = new ValueStateDescriptor<>("count", Long.class);

    var ttl = StateTtlConfig
        .newBuilder(Duration.ofDays(7))
        .setUpdateType(UpdateType.OnReadAndWrite)
        .build();
    descriptor.enableTimeToLive(ttl);

    countState = getRuntimeContext().getState(descriptor);
    processedEvents = getRuntimeContext().getMetricGroup().counter("processed_events");
  }

  @Override
  public void processElement(Event event, Context ctx, Collector<CountResult> out)
      throws Exception {
    var current = countState.value();
    var newCount = (current == null ? 0L : current) + 1;
    countState.update(newCount);
    processedEvents.inc();

    out.collect(new CountResult(event.userId(), newCount));
  }
}

Timer-Based Session (ListState)

// state: O(events per session) per key — ListState buffers the whole session
public final class SessionProcessor extends KeyedProcessFunction<String, Event, Session> {

  private ListState<Event> bufferState;
  private ValueState<Long> timerState;

  private static final long SESSION_GAP = Duration.ofMinutes(30).toMillis();

  @Override
  public void open(OpenContext ctx) {
    var bufferDescriptor = new ListStateDescriptor<>("buffer", Event.class);
    bufferState = getRuntimeContext().getListState(bufferDescriptor);

    var timerDescriptor = new ValueStateDescriptor<>("timer", Long.class);
    timerState = getRuntimeContext().getState(timerDescriptor);
  }

  @Override
  public void processElement(Event event, Context ctx, Collector<Session> out)
      throws Exception {
    bufferState.add(event);

    // cancel previous timer
    var previousTimer = timerState.value();
    if (previousTimer != null) {
      ctx.timerService().deleteEventTimeTimer(previousTimer);
    }

    // set new timer
    var newTimer = event.timestamp() + SESSION_GAP;
    ctx.timerService().registerEventTimeTimer(newTimer);
    timerState.update(newTimer);
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<Session> out)
      throws Exception {
    // RocksDB ListState.get() deserializes the ENTIRE list on every timer fire —
    // bound session size or this pattern melts under high-volume keys
    var events = new ArrayList<Event>();
    bufferState.get().forEach(events::add);

    if (!events.isEmpty()) {
      out.collect(Session.fromEvents(ctx.getCurrentKey(), events));
      bufferState.clear();
      timerState.clear();
    }
  }
}

Broadcast State

Dynamic configuration, rule engines, feature flags: broadcast a control stream to all parallel instances and connect it with the keyed event stream.

WHEN rules or config must change without redeploy → references/broadcast-state.md (KeyedBroadcastProcessFunction pattern).

Custom Metrics

Register business-level counters and gauges in open() via getRuntimeContext().getMetricGroup() (see EventCounter above for the minimal form).

WHEN adding production observability to operators → references/metrics.md (counters, gauges, latency tracking with DLQ routing).

Async I/O

NEVER block in processElement — external lookups go through AsyncDataStream.unorderedWait(stream, fn, timeout, capacity).

WHEN enriching against HTTP/DB services → references/async-io.md (RichAsyncFunction, timeout/capacity sizing, error completion).

Kafka Integration

Kafka Source

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(bootstrapServers)
    .setTopics(topic)
    .setGroupId(groupId)
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> raw = env
    .fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source")
    .uid("kafka-source");

// Parse with Result, assign watermarks after — see "Side Output DLQ Pattern" above

Kafka Sink (Exactly-Once)

KafkaSink<CountResult> sink = KafkaSink.<CountResult>builder()
    .setBootstrapServers(bootstrapServers)
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic(outputTopic)
        .setValueSerializationSchema(new JsonSerializationSchema<>())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("flink-job-" + jobName) // unique per job, mandatory
    // MUST exceed max checkpoint duration + max restart duration, or Kafka
    // expires the uncommitted transaction and data is LOST
    .setProperty("transaction.timeout.ms", "900000")
    .build();

stream.sinkTo(sink).uid("kafka-sink");

Rule: Consumers of an EXACTLY_ONCE output topic must set isolation.level=read_committed, or they read uncommitted duplicates during Flink restarts.

Key Salting for Skew

Hot keys overload single subtasks. Salt the key, pre-aggregate per salted key, then re-aggregate on the original key.

WHEN a few keys dominate traffic → references/skew.md (two-phase salted aggregation).

Testing

TDD anchor: red → green → vacuous mutation check. Regression test lands before the fix.

• Unit-test stateful operators with KeyedOneInputStreamOperatorTestHarness — never instantiate a KeyedProcessFunction directly; timers, watermarks, and TTL exist only inside the harness.
• Fire event-time timers by advancing the watermark past the registered timer: harness.processElement(...) then harness.processWatermark(...).
• Integration: MiniClusterWithClientResource from flink-test-utils.

WHEN writing operator tests → references/testing.md (dependencies, harness setup, full example).

Anti-Patterns

Five are one-liners — the owning section above carries the code:

Timer Explosion

// BROKEN: one timer per event = millions of timers
@Override
public void processElement(Event e, Context ctx, Collector<Out> out) {
  ctx.timerService().registerEventTimeTimer(e.timestamp() + delay);
}

// FIXED: one timer per key, coalesce events (see SessionProcessor)

Blocking I/O in processElement()

// BROKEN: blocks entire task, kills throughput
@Override
public void processElement(Event e, Context ctx, Collector<Out> out) {
  var result = httpClient.send(request, BodyHandlers.ofString()); // DON'T
  out.collect(transform(e, result));
}

// FIXED: use AsyncDataStream (see Async I/O section)

Breaking Savepoint Compatibility

Savepoint-breaking changes:
- Removing or changing operator UID
- Changing state serializer without migration
- Changing parallelism of keyed operations (sometimes)

Safe changes:
- Adding new operators with new UIDs
- Changing operator logic (same state schema)
- Changing parallelism of non-keyed operations

Code Review Checklist

• Versions: Flink 2.x APIs only, connector suffix matches Flink minor (see Version Baseline)
• UIDs: Every stateful operator has stable uid()
• Records: Data types are records, not mutable POJOs
• Result<T>: Deserialization and validation use Result, errors route to DLQ
• Time semantics: Event time with appropriate watermark strategy
• Late data: Explicit handling (allowed lateness or side output)
• Idle sources: .withIdleness() configured
• State TTL: Configured to prevent unbounded growth
• State size: Per-key state growth annotated on stateful operators
• ListState: Used instead of ValueState<List<T>>
• State backend: RocksDB + incremental for production
• Checkpointing: Externalized, appropriate interval
• Exactly-once Kafka: transaction.timeout.ms sized, consumers read_committed
• Broadcast state: Dynamic config uses broadcast, not hardcoded
• Metrics: Custom counters/gauges for business-level observability
• No blocking I/O: Use AsyncDataStream for external calls
• DLQ: All parsing/validation failures route to dead letter sink
• Tests: Operator harness tests for state/timer logic (TDD: red first)
• Parallelism: Matches source partitions