{"id":6161,"date":"2026-10-24T16:33:59","date_gmt":"2026-10-24T16:33:59","guid":{"rendered":"https:\/\/dalelane.co.uk\/blog\/?p=6161"},"modified":"2026-10-24T16:33:59","modified_gmt":"2026-10-24T16:33:59","slug":"how-flinks-kafka-connector-commits-offsets-to-kafka","status":"publish","type":"post","link":"https:\/\/dalelane.co.uk\/blog\/?p=6161","title":{"rendered":"How Flink’s Kafka connector commits offsets to Kafka"},"content":{"rendered":"

An unnecessarily deep dive into how Flink commits Kafka consumer offsets.<\/strong> <\/p>\n

A pull request of mine<\/a> was merged into this week<\/a>. 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. <\/p>\n

How consumer offsets are used by Flink<\/h2>\n

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. <\/p>\n

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. <\/p>\n

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.<\/p>\n

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<\/a>. <\/p>\n

Lag that wasn’t going away<\/h2>\n

FLINK-33484<\/a> reported a problem with a Flink job consuming a transactional topic with isolation.level=read_committed<\/code>. The job had read everything on the topic, but the reported lag according to Kafka was stuck at 1, not 0. <\/p>\n

A plain Kafka Java consumer reading the same data from the same topic reported no lag. Flink wasn’t committing the right offset.<\/p>\n

Every time a record is handed on to the rest of the job, the split’s position was updated<\/a>:<\/p>\n

splitState.setCurrentOffset(consumerRecord.offset() + 1);<\/pre>\n
The next offset to read is the one after the last record. <\/p>\n
That’s the right thing to do for a non-transactional, uncompacted topic, where every offset holds a record. <\/p>\n
<\/p>\n
It’s not always the right thing to do.<\/p>\n
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.<\/p>\n
<\/p>\n
This is handled by the Kafka consumer. Transaction markers aren’t given to the application. <\/p>\n
And if the application only asks for committed records (read_committed<\/code>), 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.<\/p>\n
Imagine a Kafka producer writes a single record in a transaction and commits it:<\/p>\n\n\n\n
offset 0<\/td>\nrecord<\/td>\ndelivered to Flink<\/td>\n<\/tr>\n
offset 1<\/td>\ncommit marker<\/td>\nread by the consumer
not delivered to Flink<\/td>\n<\/tr>\n<\/table>\n
The topic partition’s log end offset (and last stable offset) is 2<\/code> because that’s where the next entry will be written.<\/p>\n
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<\/em>).<\/p>\n
The Kafka consumer has actually read past the transaction marker, so its position is 2.<\/p>\n
But at the next checkpoint, Flink committed an offset of 1<\/code>. <\/p>\n
Any Kafka tooling set up to monitor Kafka consumer progress compared that with a log end of 2, and reported a lag of 1<\/strong>.<\/p>\n
<\/p>\n
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.<\/p>\n
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. <\/p>\n
That was FLINK-33484<\/a> that I picked up a little while ago.<\/p>\n
<\/p>\n
My first fix<\/h2>\n
How many “invisible” entries are there after a record, that Flink doesn’t see? <\/p>\n
For a plain topic with no transactions, maybe none.<\/p>\n
<\/p>\n
For a topic with a single transaction, maybe one.<\/p>\n
<\/p>\n
For a topic where a producer has aborted its most recent transaction, maybe four. <\/p>\n
<\/p>\n
For a topic where multiple transactional producers are sending events at the same time, maybe two.<\/p>\n
<\/p>\n
You get the idea. <\/p>\n
The point is, Flink’s KafkaRecordEmitter<\/a> that was adding 1<\/a> can’t actually know how many invisible entries follow a record. Only the Kafka consumer knows that.<\/p>\n
A Kafka consumer does know where it is, though. <\/p>\n
consumer.position(tp)<\/code> returns the offset of the next entry<\/a> it will read, and that accounts for any markers and aborted records it skipped. <\/p>\n
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.<\/p>\n
<\/p>\n
For my first fix attempt<\/a> I only used the consumer’s position when I thought it was safe. My changes were in KafkaPartitionSplitReader<\/a>, the class that owns the KafkaConsumer<\/code> and runs on the split fetcher thread:<\/p>\n
    \n
  1. At the end of each fetch()<\/code><\/a>, I recorded the offset of the last record handed to the reader for each partition<\/a>.<\/li>\n
  2. When a checkpoint completes<\/a> and the offsets are about to be committed, reconcileOffsetsToCommit<\/code> checks each partition<\/a>. If the checkpointed offset is exactly the last fetched offset + 1<\/a>, 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<\/a>.<\/li>\n<\/ol>\n
    That lastFetchedOffset + 1<\/code> check was the main idea, to stop the committed offset skipping over records that have been fetched but not yet checkpointed.<\/p>\n
    if checkpoint offset == last fetched offset + 1
    \nthen commit the consumer position<\/p><\/blockquote>\n
    If we’re caught up…<\/p>\n
    <\/p>\n
    1 == 0 + 1<\/code><\/p>\n
    Nothing is waiting in the queue, so commit the consumer position: 2<\/code><\/p>\n
    If there are still fetched records in the queue…<\/p>\n
    <\/p>\n
    2 != 2 + 1<\/code><\/p>\n
    Record 2 isn’t in the checkpoint, so commit the checkpoint’s offset: 2<\/code><\/p>\n
    It also skips partitions no longer assigned to the consumer, and bounded splits that are still running<\/a>.<\/p>\n
    This fix worked for a variety of scenarios that I came up with<\/a>, 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. <\/p>\n
    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<\/a> at commit time. But recording it during the fetch<\/a> avoided an extra call to the consumer that could block. Following a review discussion on my approach<\/a>, I added another test scenario showing what would happen on an idle topic where no more records come after a transaction marker<\/a>. <\/p>\n
    <\/p>\n\n\n\n\n\n
    poll<\/th>\nrecords returned<\/th>\nemitter runs<\/th>\nsplit’s offset<\/th>\nconsumer position<\/th>\n<\/tr>\n
    poll 1<\/td>\nrecord 0<\/td>\nyes<\/td>\n1<\/td>\n1<\/td>\n<\/tr>\n
    poll 2<\/td>\nrecord 1<\/td>\nyes<\/td>\n2<\/td>\n2<\/td>\n<\/tr>\n
    poll 3<\/td>\nnone (marker)<\/td>\nno<\/td>\n2<\/td>\n3<\/td>\n<\/tr>\n<\/table>\n
    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. <\/p>\n
    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.<\/p>\n
    At that point, I thought I’d fixed it. But reviews<\/a> showed that I’d missed a few related bugs. <\/p>\n
    Flink and Kafka disagreeing about the committed offset<\/h2>\n
    Flink has its own metrics, including metrics from the Kafka connector<\/a>. That has a committedOffset<\/code> metric with the last successfully committed offset to Kafka<\/a> for each partition. <\/p>\n
    A reviewer pointed out that my fix made Flink contradict itself<\/a>. The commit callback ignored the offsets passed to it<\/a>, and recorded the offsets that had been requested<\/a> 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<\/a>, they weren’t the same. Kafka would say the committed offset was 2 while Flink’s own metric said 1.<\/p>\n
    I had another go, and updated the fix so the callback records the offsets that were actually committed<\/a>, so Flink’s metrics and Kafka committed offsets now agree.<\/p>\n
    Partitions that never deliver a record<\/h2>\n
    A reviewer pointed out another gap<\/a>. My reconcile check depended on a last-fetched offset<\/a>. But what about partitions where nothing has ever been delivered?<\/p>\n
    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()<\/code> before anything was written to the partition:<\/p>\n
    <\/p>\n
    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.<\/p>\n
    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<\/a> reported. <\/p>\n
    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.<\/p>\n
    I tweaked my check<\/a> so that a partition with no fetched records can still be advanced to the consumer position<\/a>. If nothing has been fetched, nothing can be waiting un-checkpointed in the queue. <\/p>\n
    The placeholder offset<\/h2>\n
    That last example assumed the split started at a real offset. A split using the default OffsetsInitializer.earliest()<\/code> doesn’t have a real offset.<\/p>\n
    EARLIEST_OFFSET<\/code> is a constant representing a split starting from the earliest offset<\/a> on a partition. It’s a placeholder, represented by -2<\/code>. <\/p>\n
    A review pointed out that I hadn’t thought of that<\/a>. The split keeps that placeholder<\/a> until the first record is emitted<\/a>. This meant Flink wouldn’t commit any offset<\/a> for this partition until a record is emitted. It was another type of “no record delivered” scenario.<\/p>\n
    My next attempt<\/a> recognised these negative offset placeholders so that I could track those partitions-without-offsets<\/a> alongside the offsets. The split reader could then safely commit the consumer position for them<\/a>, because it couldn’t have fetched any records that the checkpoint didn’t include. <\/p>\n
    <\/p>\n
    Before:<\/p>\n