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.
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.
Practice what you learned
1. Why does Kafka pair particularly well with Flink's execution model?
2. What mechanism does KafkaSink use to implement EXACTLY_ONCE delivery?
3. What consumer setting is required to correctly read from a topic written by an EXACTLY_ONCE KafkaSink?
4. What happens if Kafka's transaction.timeout.ms is shorter than Flink's checkpoint interval?
5. What role does a Schema Registry play in a Flink-Kafka pipeline?
Was this page helpful?
You May Also Like
The Table API and Flink SQL
Learn how Flink's Table API and SQL layer let you express streaming and batch pipelines declaratively on top of the DataStream engine.
Scaling and Parallelism
Learn how Flink parallelizes work across task slots and how to scale jobs up or down without losing correctness.
Deploying Flink on Kubernetes
Understand the native Kubernetes deployment modes for Flink, how the Flink Kubernetes Operator manages job lifecycle, and key production configuration.
Related Reading
Related Study Notes in Programming
Browse all study notesApache Spark Study Notes
Programming · 30 topics
ProgrammingHadoop Study Notes
Programming · 30 topics
ProgrammingSnowflake Study Notes
Programming · 30 topics
ProgrammingApache Airflow Study Notes
Programming · 30 topics
Programmingdbt (Data Build Tool) Study Notes
Programming · 30 topics
ProgrammingPowerShell Study Notes
Programming · 30 topics