Flink with Kafka

Analogy categoryAnalogies

Why Kafka Is Flink's Most Common Companion

Kafka provides Flink with a durable, replayable, partitioned log, which matches Flink's own execution model closely: Kafka partitions map naturally onto parallel source subtasks, Kafka's offset-based replay lets Flink restart from a checkpoint without data loss, and Kafka's retention window gives Flink time to recover from failures without the source discarding data. This alignment is why nearly every production Flink deployment uses Kafka (or a Kafka-compatible log like Redpanda or Amazon MSK) as its primary ingestion layer.

🏏

Cricket analogy: It's like a stadium's ball-by-ball commentary archive — because every delivery is logged with a sequence number, a broadcaster who loses connection can resume exactly from ball 247 rather than guessing where they left off, just as Flink resumes from a Kafka offset.

Configuring the Kafka Source and Sink

The modern KafkaSource builder (replacing the deprecated FlinkKafkaConsumer) lets you set bootstrap servers, topic(s) or a topic pattern, a deserializer, and a starting offset strategy such as OffsetsInitializer.earliest(), .latest(), or .committedOffsets(). On the write side, KafkaSink supports three delivery guarantees — NONE, AT_LEAST_ONCE, and EXACTLY_ONCE — with exactly-once implemented via Kafka's transactional producer API, meaning Flink writes records inside a Kafka transaction that only commits when the corresponding checkpoint completes.

🏏

Cricket analogy: Choosing a starting offset is like a substitute umpire joining mid-match and deciding whether to review only from the current over (latest), from ball one (earliest), or from the last confirmed decision (committed) before making calls.

java

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

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

DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-orders");
stream.sinkTo(sink);

End-to-End Exactly-Once Semantics

Achieving true end-to-end exactly-once requires all three legs to cooperate: Kafka source offsets are stored as part of Flink's checkpoint state (not committed independently to Kafka), Flink's internal state is snapshotted via the checkpoint barrier mechanism, and the KafkaSink pre-commits its transaction during the checkpoint and only finalizes the commit once the checkpoint is confirmed complete — this two-phase-commit protocol is why EXACTLY_ONCE delivery requires checkpointing to be enabled and requires downstream Kafka consumers to read with isolation.level=read_committed to avoid seeing uncommitted, potentially-rolled-back data.

🏏

Cricket analogy: It's like a third umpire's DRS decision — the on-field call is provisional (pre-committed) until the TV replay confirms it, and only then does the scoreboard (Kafka log) officially update; viewers with the replay feed on 'confirmed only' mode are like read_committed consumers.

Set isolation.level=read_committed on any downstream Kafka consumer reading a topic written by an EXACTLY_ONCE KafkaSink; otherwise consumers may read uncommitted records from transactions that are later aborted during a Flink failure recovery.

EXACTLY_ONCE Kafka sinks hold open transactions between checkpoints. If your checkpoint interval is long and a consumer's transaction.timeout.ms is shorter than that interval, Kafka will abort the transaction before Flink commits it, silently dropping data — always align these two settings.

Schema Evolution with Kafka Topics

Because Kafka stores raw bytes, schema management is external to the log itself — most production setups pair Flink with a Schema Registry (Confluent or Apicurio) and use Avro or Protobuf formats so that both producers and Flink's deserializer validate against a shared, versioned schema, allowing backward-compatible field additions without breaking already-running consumers.

🏏

Cricket analogy: It's like the ICC maintaining a central rulebook that all member boards reference — a new law addition (like the concussion substitute rule) is versioned and rolled out so old matches' scorecards remain valid under the rules they were played under.

Kafka's partitioned, replayable log model aligns naturally with Flink's parallel, checkpoint-based recovery model.
KafkaSource replaces the deprecated FlinkKafkaConsumer and supports configurable starting offsets (earliest, latest, committedOffsets).
KafkaSink supports NONE, AT_LEAST_ONCE, and EXACTLY_ONCE delivery guarantees.
EXACTLY_ONCE relies on Kafka transactions coordinated with Flink's checkpoint barriers via two-phase commit.
Downstream consumers must use isolation.level=read_committed to avoid seeing data from aborted transactions.
Checkpoint interval and Kafka's transaction.timeout.ms must be aligned to avoid silent data loss from transaction expiry.
Schema Registry with Avro/Protobuf enables backward-compatible schema evolution between producers and Flink consumers.