Sources: Getting Data Into Flink
Modern Flink jobs read data using the unified Source interface introduced by FLIP-27, which splits the work of ingestion into a SplitEnumerator running on the JobManager that discovers and assigns work units (splits), and one or more SourceReaders running on TaskManagers that actually read records from those splits, such as individual Kafka partitions or file shards. This design unifies bounded and unbounded reading under one API and lets connectors like KafkaSource handle dynamic partition discovery, checkpointed offsets, and backpressure-aware reading without custom code in the job itself.
Cricket analogy: The SplitEnumerator is like a tournament scheduler assigning specific matches to specific stadiums, while each stadium's ground staff (SourceReader) is responsible for actually running the fixtures assigned to it.
Sinks: Getting Data Out
The counterpart Sink interface defines how records leave a Flink job, typically through a SinkWriter that batches and writes records to an external system, optionally coordinated by a Committer and GlobalCommitter for connectors that support two-phase commit, such as KafkaSink configured with DeliveryGuarantee.EXACTLY_ONCE. In that mode, writes are staged inside a transaction tied to a checkpoint, and only committed once Flink confirms the checkpoint has completed successfully across the whole job, which is what allows exactly-once output semantics even after a failure and restart.
Cricket analogy: A two-phase commit sink is like a third umpire's decision that stays 'pending' on the big screen until confirmed by the review process, only becoming official once every check has passed — not the instant the on-field umpire raises a finger.
KafkaSource<String> source = KafkaSource.<String>builder()
.setBootstrapServers("broker1:9092,broker2:9092")
.setTopics("orders")
.setGroupId("flink-orders-consumer")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
KafkaSink<String> sink = KafkaSink.<String>builder()
.setBootstrapServers("broker1:9092,broker2:9092")
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic("orders-enriched")
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "orders-source")
.map(order -> enrich(order))
.sinkTo(sink);
env.execute("Kafka To Kafka Enrichment");Delivery Guarantees
The end-to-end delivery guarantee of a Flink pipeline depends on the combination of the source's ability to replay records from a checkpointed offset, Flink's own checkpointing to make operator state consistent, and the sink's own guarantee, which ranges from at-most-once (no retry on failure), through at-least-once (retries can produce duplicates), to exactly-once (achieved via idempotent writes or two-phase commit transactions). A pipeline is only as strong as its weakest link, so pairing an exactly-once-capable source like Kafka with a non-transactional, non-idempotent sink still only yields at-least-once semantics overall.
Cricket analogy: It's like a run being counted correctly only if both the batsmen actually complete the run and the umpire confirms it — a fast start with a run-out at the far end still means the run doesn't count, no matter how good the start was.
Two-phase commit sinks rely on Flink's checkpoint barriers to know when it's safe to commit a transaction: the sink pre-commits (stages) writes as part of each checkpoint, and only actually commits them once the checkpoint has been acknowledged as complete by every operator in the job, tying external write durability directly to Flink's own checkpointing protocol.
Custom Connectors
When no existing connector fits a bespoke system, Flink allows implementing a custom Source by providing a SplitEnumerator that discovers and assigns splits (e.g. partitions of a proprietary message queue) and a SourceReader that reads records from an assigned split and reports its progress for checkpointing, or a custom Sink by implementing a SinkWriter and, if exactly-once is required, a Committer that finalizes staged writes only after a successful checkpoint. This is more involved than using an existing connector, but it lets any bespoke external system participate correctly in Flink's checkpointing and fault-tolerance model.
Cricket analogy: Writing a custom connector is like a franchise building its own scouting network from scratch in a country with no existing cricket board infrastructure, rather than simply plugging into the well-established BCCI pipeline.
Under at-least-once delivery, retries after a failure can cause the same record to be written more than once. If your sink is not naturally idempotent — for example, a plain INSERT into a relational table without a unique constraint on a business key — duplicate writes will silently corrupt downstream data even though the pipeline appears healthy.
- Flink's modern Source interface (FLIP-27) splits ingestion into a SplitEnumerator and per-subtask SourceReaders.
- Sinks write via a SinkWriter, optionally coordinated by a Committer for two-phase commit exactly-once writes.
- End-to-end delivery guarantees depend on the weakest link among source replay, checkpointing, and sink behavior.
- Two-phase commit sinks tie transaction commits directly to Flink's checkpoint completion.
- Custom Sources and Sinks let bespoke external systems participate correctly in checkpointing.
- At-least-once delivery can produce duplicates unless the sink is idempotent.
- Choosing the right delivery guarantee requires understanding both source and sink capabilities together.
Practice what you learned
1. What is the role of the SplitEnumerator in Flink's Source interface?
2. How does a two-phase commit sink achieve exactly-once output semantics?
3. Why can a pipeline with an exactly-once-capable Kafka source still only achieve at-least-once overall?
4. What must a custom Sink implement to support exactly-once semantics?
5. What risk does at-least-once delivery carry for a non-idempotent sink?
Was this page helpful?
You May Also Like
DataStream Basics
An introduction to Flink's DataStream API, the core abstraction for processing unbounded and bounded streams of data in real time.
Keyed Streams and keyBy
Partitioning a DataStream by key using keyBy() to enable per-key state and correct windowed aggregations.
Serialization in Flink
How Flink serializes records for network transfer, state storage, and checkpointing, and why type information matters for performance.
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