An unnecessarily deep dive into how Flink commits Kafka consumer offsets.
A pull request of mine was merged into this week. A ridiculous number of sketches and diagrams went into the commits to get the idea straight in my head, so in this post I tidied them up a bit to share.
How consumer offsets are used by Flink
If you run a Flink job that reads from Kafka, you’ve probably seen its consumer group in your Kafka tooling like any other Kafka consumer: committed offsets for each partition, and a lag figure showing how far behind the log end the job is.
Flink doesn’t need to depend on those committed offsets for its own work. The unit of work for a Flink source is a “split”. For Kafka, a split is a partition and the offset to read from (and for a bounded split, the offset to stop reading at). This is stored in Flink state. Each source reader holds a split for every partition it reads, and each split records the next Kafka offset to read.
When Flink takes a checkpoint, each split is saved in the checkpoint, and contains those offsets. If the job fails, or you restart it from a savepoint, the source picks up from the offsets in the checkpoint. That’s how Flink can offer exactly-once processing: the offsets and the rest of the job’s state are captured together, in the same consistent snapshot.
Typically, the Kafka source reader commits the checkpointed offsets to Kafka once a checkpoint has completed. Although that is not necessary for the Flink job, it’s useful for lag dashboards and alerting, in the same way as any other Kafka consumer, and helps to understand how the job is doing.
Lag that wasn’t going away
FLINK-33484 reported a problem with a Flink job consuming a transactional topic with isolation.level=read_committed. The job had read everything on the topic, but the reported lag according to Kafka was stuck at 1, not 0.
A plain Kafka Java consumer reading the same data from the same topic reported no lag. Flink wasn’t committing the right offset.
Every time a record is handed on to the rest of the job, the split’s position was updated:
splitState.setCurrentOffset(consumerRecord.offset() + 1);
The next offset to read is the one after the last record.
That’s the right thing to do for a non-transactional, uncompacted topic, where every offset holds a record.
It’s not always the right thing to do.
When a Kafka producer commits or aborts a transaction, the broker writes a transaction marker into each partition the transaction touched. That marker takes up an offset in the log like any other entry.
This is handled by the Kafka consumer. Transaction markers aren’t given to the application.
And if the application only asks for committed records (read_committed), records from aborted transactions are skipped in the same way. The consumer reads past them, and its position moves forward, but no record comes out for them.
Imagine a Kafka producer writes a single record in a transaction and commits it:
| offset 0 | record | delivered to Flink |
| offset 1 | commit marker | read by the consumer not delivered to Flink |
The topic partition’s log end offset (and last stable offset) is 2 because that’s where the next entry will be written.
Flink emitted the record at offset 0, and set the split’s current offset to 1 (because it added 1 to the offset it’s emitted).
The Kafka consumer has actually read past the transaction marker, so its position is 2.
But at the next checkpoint, Flink committed an offset of 1.
Any Kafka tooling set up to monitor Kafka consumer progress compared that with a log end of 2, and reported a lag of 1.
On a busy topic, you’d barely notice. The next record arrives, gets emitted, and the committed offset jumps over the marker. The committed offset was always slightly behind, but only until new data arrives.
On a quiet or bursty topic, “until new data arrives” can be a long time. The job was fully caught up, but every checkpoint committed the same stale offset, and Kafka monitoring reported a lag that didn’t exist.
That was FLINK-33484 that I picked up a little while ago.
My first fix
How many “invisible” entries are there after a record, that Flink doesn’t see?
For a plain topic with no transactions, maybe none.
For a topic with a single transaction, maybe one.
For a topic where a producer has aborted its most recent transaction, maybe four.
For a topic where multiple transactional producers are sending events at the same time, maybe two.
You get the idea.
The point is, Flink’s KafkaRecordEmitter that was adding 1 can’t actually know how many invisible entries follow a record. Only the Kafka consumer knows that.
A Kafka consumer does know where it is, though.
consumer.position(tp) returns the offset of the next entry it will read, and that accounts for any markers and aborted records it skipped.
But I couldn’t just commit the consumer’s position. The consumer fetches batches of records from Kafka, ahead of what Flink has processed so far. Records can be sitting in the fetcher’s queue, waiting to be emitted. When Flink takes a checkpoint, the checkpoint doesn’t include the records that Flink hasn’t processed yet. The offset committed to Kafka shouldn’t either.
For my first fix attempt I only used the consumer’s position when I thought it was safe. My changes were in KafkaPartitionSplitReader, the class that owns the KafkaConsumer and runs on the split fetcher thread:
- At the end of each
fetch(), I recorded the offset of the last record handed to the reader for each partition. - When a checkpoint completes and the offsets are about to be committed,
reconcileOffsetsToCommitchecks each partition. If the checkpointed offset is exactly the last fetched offset + 1, then the checkpoint includes every record the consumer has delivered. Anything between that and the consumer’s position must be entries that were never going to be delivered. In that case, it’s safe to advance the offset to the consumer position.
That lastFetchedOffset + 1 check was the main idea, to stop the committed offset skipping over records that have been fetched but not yet checkpointed.
if checkpoint offset == last fetched offset + 1
then commit the consumer position
If we’re caught up…
1 == 0 + 1
Nothing is waiting in the queue, so commit the consumer position: 2
If there are still fetched records in the queue…
2 != 2 + 1
Record 2 isn’t in the checkpoint, so commit the checkpoint’s offset: 2
It also skips partitions no longer assigned to the consumer, and bounded splits that are still running.
This fix worked for a variety of scenarios that I came up with, such as single-record transactions, multi-record transactions, committed transactions followed by aborted transactions, several transactions in a row, and interleaved transactions from two producers.
Part of the motivation for this approach was in deciding what to commit as late as possible, at the time of committing the offset. At first, I did this by asking the consumer for its position at commit time. But recording it during the fetch avoided an extra call to the consumer that could block. Following a review discussion on my approach, I added another test scenario showing what would happen on an idle topic where no more records come after a transaction marker.
| poll | records returned | emitter runs | split’s offset | consumer position |
|---|---|---|---|---|
| poll 1 | record 0 | yes | 1 | 1 |
| poll 2 | record 1 | yes | 2 | 2 |
| poll 3 | none (marker) | no | 2 | 3 |
If you decide what to commit as records are emitted, then because poll 3 has no records, the emitter is never called and the offset stays at 2. The commit will lag until more data arrives.
By deciding when the commit happens, it can use the position from the most recent poll, whether or not it returned any records. The committed offset can be 3, with no lag.
At that point, I thought I’d fixed it. But reviews showed that I’d missed a few related bugs.
Flink and Kafka disagreeing about the committed offset
Flink has its own metrics, including metrics from the Kafka connector. That has a committedOffset metric with the last successfully committed offset to Kafka for each partition.
A reviewer pointed out that my fix made Flink contradict itself. The commit callback ignored the offsets passed to it, and recorded the offsets that had been requested instead. That had always been fine, because the two were always the same. Once my split reader started changing the offsets on their way to Kafka, they weren’t the same. Kafka would say the committed offset was 2 while Flink’s own metric said 1.
I had another go, and updated the fix so the callback records the offsets that were actually committed, so Flink’s metrics and Kafka committed offsets now agree.
Partitions that never deliver a record
A reviewer pointed out another gap. My reconcile check depended on a last-fetched offset. But what about partitions where nothing has ever been delivered?
For example, a partition holding only an aborted transaction (aborted records at offsets 0 to 5, the abort marker at 6), read by a split whose starting offset is 0, such as a job started with OffsetsInitializer.latest() before anything was written to the partition:
In this case, the Kafka source reader is fully caught up on the topic, but in every checkpoint Flink commits offset 0, so from Kafka’s perspective the consumer has a lag of 7 until a Kafka producer writes a committed record to move it on.
That wasn’t one of the scenarios I’d come up with, but it was a worse version of the original bug than FLINK-33484 reported.
The reader is fully caught up, but every checkpoint commits 0, and monitoring would report a lag of 7 until someone writes a committed record.
I tweaked my check so that a partition with no fetched records can still be advanced to the consumer position. If nothing has been fetched, nothing can be waiting un-checkpointed in the queue.
The placeholder offset
That last example assumed the split started at a real offset. A split using the default OffsetsInitializer.earliest() doesn’t have a real offset.
EARLIEST_OFFSET is a constant representing a split starting from the earliest offset on a partition. It’s a placeholder, represented by -2.
A review pointed out that I hadn’t thought of that. The split keeps that placeholder until the first record is emitted. This meant Flink wouldn’t commit any offset for this partition until a record is emitted. It was another type of “no record delivered” scenario.
My next attempt recognised these negative offset placeholders so that I could track those partitions-without-offsets alongside the offsets. The split reader could then safely commit the consumer position for them, because it couldn’t have fetched any records that the checkpoint didn’t include.
Before:
- Split state
- offset = EARLIEST_OFFSET
placeholder (-2) until record is emitted
- offset = EARLIEST_OFFSET
- snapshotState
- negative offset
so the split is skipped
- negative offset
- no records fetched – nothing to commit
- Kafka
- no offsets are committed for this partition
Now:
- Split state
- offset = EARLIEST_OFFSET
placeholder (-2) until record is emitted
- offset = EARLIEST_OFFSET
- snapshotState
- records the partition as a partition without an offset
- no records fetched – commit the consumer position
- Kafka
- committed offset = 7
Moving backwards
Another review comment and another problem scenario spotted.
Consider a topic with a transactional record at offset 0 with a commit marker at offset 1.
The reader emits the record at offset 0
The consumer moves to position 2, past the transaction marker.
No offsets are committed to Kafka yet.
Checkpoint 1 snapshots offset 1.
Reconciliation advances it.
Offset 2 is committed to Kafka.
Checkpoint 2 snapshots offset 1 (again).
The offset committed to Kafka is still 2
The partition was idle, but before the checkpoint completes…
A new record arrives at offset 2.
The record is fetched.
The raw checkpointed offset, 1, would be committed to Kafka
The offset committed to Kafka went backwards, from 2 to 1. This is the sort of thing that if it happened for real you’d hope would trigger some alert somewhere because that’s a weird sign.
The split reader now stores the offset it last committed for each assigned partition, so it can leave a partition out of a commit rather than commit a lower offset.
How it works now
Those are the kinds of things that I changed along the way – there were others, but let me skip to the end and describe where I ended up.
This is what happens to a committed offset now, from fetch to Kafka.
While fetching
On the split fetcher thread, KafkaPartitionSplitReader.fetch() polls the consumer.
For each partition it records:
lastKnownPositions: the consumer’s position at the end of the fetch, which accounts for markers and aborted records it has read pastlastFetchedOffsets: offset of the last record it handed to the reader
KafkaRecordEmitter sets each split’s current offset to the offset after the record it emits.
When Flink takes a checkpoint
KafkaSourceReader.snapshotState collects each split’s current offset to commit.
Splits that are still at a starting placeholder such as EARLIEST_OFFSET have no real offset yet, so their partitions are now recorded separately for that checkpoint.
When a checkpoint completes
The reader enqueues a commit task on the split fetcher, with the checkpoint’s offsets and its partitions with no offset.
The commit task runs on the fetcher thread (the only thread that touches the consumer) and reconciles them before committing:
- For a partition with a checkpointed offset, if the checkpoint covers every record fetched so far (or none have been fetched) and the consumer position is further ahead, the offset is advanced to the consumer position
- For a partition with no offset, the consumer position is committed, as long as no records have been fetched for it
- Partitions that are no longer assigned, and bounded splits that are still running, aren’t moved to the consumer position
- Any partition whose offset would be lower than the one last committed is left out, so while a partition stays assigned to a reader, its committed offset doesn’t move backwards. The offsets being committed are recorded as soon as the commit is sent.
If nothing is left to commit, the callback is called without a round trip to Kafka.
When a bounded split finishes
It’s committed with the next checkpoint. I changed this to use its stopping offset (when that’s a concrete offset) instead of the offset after its last emitted record, unless the split never left its starting placeholder.
When the Kafka commit completes
The callback records Flink’s committedOffset metric from the offsets Kafka actually received, so Flink’s metrics and Kafka’s tooling agree.
If partitions are unassigned
Their tracking state is removed.
Phew
None of my changes altered Flink’s processing guarantees. I didn’t change what goes into a checkpoint. This has all been about the offsets committed to Kafka (and Flink’s committedOffset metric so they are consistent).
When the checkpoint covers everything the consumer has delivered, the committed offsets for unbounded splits now match the consumer’s position as of its most recent poll.
I hope.
Summary
I thought I was just picking up a simple off-by-one fix for Flink jobs on transactional topics. I think I even described this as “a quick fix” when I started.
A series of reviews raised questions that were each a variation on the same theme: what if the partition has delivered nothing at all? What if the split hasn’t got a real offset yet? What if a record arrives between the checkpoint and the commit? What if the split is bounded?
The idea at the core of the fix didn’t change much from my first commit, but pretty much everything around it did. 🙂
This isn’t finished!
I’m not claiming that committed offsets are now right in every case. My pull request was about which offset gets committed when a commit happens. It doesn’t change what triggers a commit, or what happens if a commit is never triggered.
There are other aspects to consider. For example, FLINK-36532 is looking at stopping a job with a savepoint. The savepoint holds the source’s offsets, but they never reach Kafka. The pull request opened for it explains that the cause is the asynchronous design described above. When a checkpoint completes, the commit is queued as a task on the split fetcher. Stop-with-savepoint closes the reader straight after the savepoint completes, and the fetcher shuts down without running the queued task. The consumer group is left at an earlier checkpoint’s offsets, or with no committed offsets at all if no earlier commit ran.
The committed offset is a side channel. It’s best effort, it’s never allowed to fail the job, and Flink’s processing guarantees don’t cover it. This means no single piece of code is responsible for getting it right. Each step in a reader’s life (a checkpoint completing, a split finishing, a reader closing, a job restarting) decides for itself what reaches Kafka. I’ve described a few of those steps here. And I’ve shown how easy it was for me to miss steps!
Thanks
Huge thanks to Aleksandr, Martijn, Efrat, Arvid for their kind patience with me on this pull request – it was their reviews that found most of what I’ve written about here, and I learned a lot about the internals of Flink’s Kafka connector from the process.
Tags: apacheflink, apachekafka, flink, kafka