demo: do not review | CC-41637: Commit the frontier offset in the S3 sink to clear phantom lag and restart re-reads

[Sangeet Mishra (sangeet259)]

Demonstration PR. Please do not review or merge. It exists to illustrate the decided-frontier approach for CC-41637.

Summary

The S3 sink currently advances the committed Kafka offset only when a file rotates and is uploaded to S3.
When records are consumed but no file is uploaded, the committed offset freezes, and monitoring shows lag that never drains even though the connector is healthy.
This change adds an opt-in mode, controlled by a new config flag commit.frontier.offset.enable that defaults to false, in which preCommit reports the frontier offset per partition instead of only the offset of the last uploaded file.
The frontier is the lowest Kafka offset of a record that has been buffered but not yet uploaded, or the consumed high-water mark when nothing is buffered.
Everything below the frontier is decided, so committing it clears the phantom lag without ever committing past a record that is still in flight.

The bug as it manifests in S3

The S3 sink buffers records in put() and only makes them durable when a file rotates and its upload commits.
The committable offset is derived from that rotation: TopicPartitionWriter.commitFiles() sets offsetToCommit = currentOffset + 1, where currentOffset is the highest offset written into the file that was just uploaded, and preCommit forwards that value through getOffsetToCommitAndReset().
Nothing else advances the committed offset.

That leaves several ordinary situations where offsets are consumed but the committed offset does not move, so the partition carries a lag that never drains on its own.

• An idle partition that receives no new records never rotates a file, so its committed offset stays wherever the last upload left it, even though the consumer has caught up to the log end.
• A run of records dropped inside put() because their value is null and behavior.on.null.values is ignore never enters a file, so it never contributes to a rotation.
• Records removed by a Filter transform upstream never reach put() at all, so the connector cannot even name those offsets, yet the framework counts them as consumed.
• A buffer that is still below the flush size, and whose partition sees no time-based or scheduled rotation, holds real records that are consumed but not yet uploaded, so their offsets are neither committed nor lost, just stuck.

In every one of these cases the consumer group has moved forward but the committed offset has not, and the difference shows up as lag that looks like a stall.

For the S3 sink there is a second consequence that matters even more than the misleading metric.
On restart the S3 sink resumes from the committed Kafka offset.
Because the committed offset froze, a restart re-reads from the frozen position and reprocesses everything above it, including the records it already dropped or filtered, which is wasted work and, for a null-drop run, a repeat of the same decision.

The frontier mechanism, mapped to S3

The fix rests on a single rule.
Never commit past a record that the connector intends to write to the end system and has not yet made durable.
Every offset at or below the committed offset must be decided, and decided means one of two things: the record was durably written, or the connector deliberately decided it will never be written, whether by dropping a null value, sending a record to the dead-letter queue, or having a Filter transform remove it before it arrived.

Rather than trying to enumerate every reason an offset became safe, which is impossible because filtered records never reach the connector, the rule is stated as a negative.
The only offsets that are not safe to commit are those whose records the connector intends to write and has not yet made durable.
Call that set of pending offsets P.
The frontier is then defined as:

frontier = P.isEmpty() ? consumedHighWaterMark : min(P)

When a record is still in flight, the frontier is the lowest such offset, and everything below it is decided.
When nothing is in flight, the frontier is the consumed high-water mark, the farthest offset the connector could safely commit.
The frontier is committed as is, not min(P) - 1, because a committed offset names the next record to consume, so committing min(P) makes the lowest pending record be re-read after a restart, which is exactly the intended recovery point.

Mapping the abstract terms to the real S3 code:

• "durable" means a record written into a file that has been uploaded and committed to S3, which is the moment commitFiles() runs and sets offsetToCommit.
• "pending", the set P, means offsets of records that have been buffered but not yet uploaded, whether they are still sitting in the in-memory buffer or have already been written into an open file that has not yet rotated.

The implementation tracks P per TopicPartitionWriter in an ordered set.
An offset enters the set in buffer(SinkRecord), which is the single place a record is accepted, so no code path can buffer a record without it becoming visible to the frontier computation.
An offset leaves the set in two ways.
When commitFiles() uploads files, every offset at or below the newly durable currentOffset is removed, because those records are now durable.
When a buffered record is instead sent to the dead-letter queue or dropped by a partition error, it is removed as well, because a drop is itself a decision and the record will never be uploaded, so it must not hold the frontier back.
The minimum of the set is then a cheap first() lookup, and preCommit uses it, falling back to the consumed high-water mark from its own argument when the set is empty.

Because the frontier is computed only from whether an offset was buffered and not yet made durable, the invisible filtered offsets and the dropped null offsets between the durable position and the frontier are stepped over without the connector ever having to name them.
If one of those offsets had actually been a real record still pending, it would be in P, and being lower than the current minimum it would have pulled the frontier down to itself, which is a contradiction, so every offset below the frontier is provably decided.

Why S3 needs strictly less than a re-seek sink

This is the point of demonstrating the frontier idea on S3 specifically.
Sinks fall into two groups by how they recover on restart.

A re-seek sink, such as Snowflake, does not resume from the committed Kafka offset.
It resumes from its own external durable state, for example a channel offset token, and it re-seeks to that token plus one, ignoring the committed Kafka offset.
For such a sink, committing the frontier fixes the phantom lag in monitoring, but it does nothing for restart on its own, because restart consults the external token rather than the committed offset.
To make restart land on the frontier, a re-seek sink has to persist the frontier alongside the commit, in the offset metadata, and read it back on restart to seek to the maximum of the durable token and that recovery floor.
That requires extra recovery machinery: a metadata floor written on every commit, and an open()-time read and re-seek that consults it.

S3 is a non-re-seek sink.
On restart it resumes directly from the committed Kafka offset.
So the moment preCommit commits the frontier, both problems are solved at once.
The lag in monitoring reflects only genuine in-flight work, and a restart resumes from the committed offset, which is already the frontier, so it re-reads the lowest pending record and skips nothing that was not already decided.
There is no metadata floor to write, no open()-time read, and no re-seek to implement.
The single change in preCommit is the entire mechanism, and the connector runs on a stock Connect runtime.
This is the contrast worth stating plainly: the same frontier idea, but S3 needs strictly less than Snowflake because it recovers from the committed offset that the frontier already produces.

Why it is data-loss-safe

Everything rests on one invariant: every consumed real record is, at every commit, either already durable in S3 or currently in the pending set P, never neither.

This holds from the order of operations in Connect and from where the pending set is populated.
The framework hands a batch to put().
put() buffers each record it accepts, and buffering is exactly where the offset enters P, before put() returns.
Only after put() returns does the framework advance the consumed position that preCommit later reads.
So by the time any commit reads the consumed high-water mark, every real record below it that the connector accepted is already in P or already uploaded.
There is no window in which a consumed real record is missing from both, and that window is precisely what a data-loss bug would require.
The frontier, being the minimum of P or the consumed edge when P is empty, therefore never names an offset above a pending real record, so committing it can never step over a record that still needs to be written.

The drop paths preserve the invariant rather than breaking it.
A record removed to the dead-letter queue or dropped by a partition error is a decision, so it correctly leaves P, and because it will never be uploaded, leaving it in P would only pin the frontier and reintroduce a stall, never cause a loss.
A record written into an open file that has not yet rotated stays in P until commitFiles() makes it durable, so a crash before rotation leaves the committed offset at or below that record and it is re-read after restart.
On a retriable write error the sink discards the writer and rebuilds it, and the framework re-seeks and re-buffers, which repopulates P from the re-read, so the pending set is self-healing.

Config flag and default-off safety

The behavior is gated by a new boolean config, commit.frontier.offset.enable, in the S3 config group, defaulting to false.
When it is false, the code path is byte-for-byte the existing one: preCommit calls getOffsetToCommitAndReset() per writer exactly as before, the pending set is never populated, and there is no behavioral or performance change.
When it is true, preCommit computes the frontier per partition as described.
Because the default is off, this change is inert for every existing connector until an operator opts in, which keeps the rollout safe and reversible by a single config edit.

Test plan

• Unit tests on TopicPartitionWriter for the pending-set lifecycle: an offset enters the set on buffer, leaves on commitFiles for everything at or below the new durable offset, and leaves on a dead-letter or partition-error drop, so that getFrontierOffset returns min(P) when records are in flight and the supplied consumed high-water mark when the set is empty.
• Unit test on S3SinkTask.preCommit that with the flag off the committed offsets are identical to the current behavior, and with the flag on the committed offset for an idle or fully drained partition advances to the consumed high-water mark from the preCommit argument, while a partition with buffered-but-not-uploaded records commits min(P).
• The worked scenario from the design note, driven end to end against a real bucket: a partition with a mix of uploaded records, null drops, filtered records, and buffered-not-uploaded records, asserting that the committed offset sits on the lowest buffered offset and ratchets forward as files rotate, reaching the consumed edge once the last buffered record is uploaded, with the stock connector visibly stuck by contrast.
• A restart test that exercises the non-re-seek recovery: with the flag on, kill the task while records are buffered but not uploaded, restart, and confirm the task resumes from the frontier, re-reads exactly the pending records, and does not reprocess the already-uploaded or already-dropped offsets below the frontier, and confirm no record is lost.
• A regression pass of the existing S3 sink test suite with the flag off to confirm the default path is unchanged.

Where the code lives

The change is three files in kafka-connect-s3.

• io/confluent/connect/s3/S3SinkConnectorConfig.java: the new commit.frontier.offset.enable config key, its ConfigDef definition in the S3 group, and the isCommitFrontierOffsetEnabled() accessor.
• io/confluent/connect/s3/TopicPartitionWriter.java: the per-partition pending set P, populated in buffer, pruned in commitFiles, and pruned on the dead-letter and partition-error drop paths, plus getFrontierOffset(consumedHighWaterMark).
• io/confluent/connect/s3/S3SinkTask.java: preCommit branches to preCommitFrontier when the flag is on, computing the frontier per partition from the pending set and the consumed position carried in its own argument.

Build status

The kafka-connect-s3 module compiles cleanly under JDK 11, including checkstyle validation.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Pull request overview

This PR adds an opt-in “frontier offset” committing mode for the S3 sink so committed Kafka offsets advance even when no file rotation/upload occurs, reducing phantom lag and avoiding unnecessary re-reads after restarts.

Changes:

• Introduces commit.frontier.offset.enable (default false) to toggle frontier-based committing.
• Tracks pending (buffered-but-not-durable) offsets per partition in TopicPartitionWriter and exposes getFrontierOffset(...).
• Updates S3SinkTask.preCommit to optionally commit the per-partition frontier instead of only the last-uploaded-file offset.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
kafka-connect-s3/src/main/java/io/confluent/connect/s3/TopicPartitionWriter.java	Adds pending-offset tracking and frontier computation; prunes pending offsets on commit and DLQ/drop paths.
kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkTask.java	Adds frontier-based preCommit branch to commit buffered frontier or consumed high-water mark.
kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java	Adds the new config key, docs, and accessor for enabling frontier commits.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[sonarqube-confluent]

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube