kafka connect: coordinator only commits offsets when greater than existing offsets

[twthorn]

Fixes #17551

More context in that ticket on the exact sequence of events.

Overall, it's possible multiple coordinators exist, and one may commit old/stale offsets still in memory, which may be out of retention, and this sequence will cause data loss.

We do a check before writing the offset.

Note: it is possible that a race condition exists (eg coordinator A reads offset n, coordinator B reads offset n & commits offset n+2, and then coordinator A commits offset n+1). However, the committed offset still never drops below n, so the worst case is reprocessing a few records (ie duplicates), not data loss.

Also add some logging that makes these scenarios much more clear (eg when a stale coordinator may exist, what the coordinators offsets are that they are committing).

[twthorn]

@laskoviymishka think you could take a look when you get a free moment? Thanks!

[laskoviymishka]

Nice, this is a real bug (#17551) and the monotonic guard is the right shape for it; stopping a stale coordinator from rewinding the committed offset is exactly what we want.

I'd hold this before merging though. The part all three review passes landed on is the new consumer.committed(partitions) call. It's a synchronous broker round-trip added to every commit cycle, and because it isn't a CommitFailedException, a transient TimeoutException propagates through doCommit() and terminates the task as non-retryable. The old path never made that RPC, so this trades one recovery bug for a new way to kill the connector on a broker blip. I'd get the last-committed value from a local cache updated after each commitSync and do the check there, which removes the round-trip entirely — details inline.

The other thing I'd want before merge is the test actually proving the guard. Right now it asserts isGreaterThanOrEqualTo(healthyWatermark), which passes even if the guard never ran, and the lastCommitted == null first-commit branch — the one that matters most for a fresh coordinator — is never exercised.

A few things I'd like to settle in this PR before merge:

• drop the per-cycle broker round-trip (local cache, or at minimum an explicit timeout + catch so we fall back to committing rather than dying)
• tighten the test to isEqualTo and add the null / first-commit case
• downgrade the skip-path log to debug so idle connectors don't flood INFO

The concurrent-coordinator TOCTOU you already call out in the description is fine to leave for a follow-up, but I'd note in the code that the window is intentionally still open so nobody assumes otherwise.

Once those are addressed, happy to take another pass and approve.

[twthorn]

@laskoviymishka ready for re-review when you get the chance. Thank you for the thorough review!

[laskoviymishka]

This is good to land, thanks for this!

left couple follow up nit's

[twthorn]

@laskoviymishka thanks for the detailed review, updated with all feedback when you get the chance for another pass, thank you!

[twthorn]

Perhaps @nssalian can you take a look at this? Already has approval from @laskoviymishka. I've addressed the remaining nits. Thanks!

[nssalian]

Just one comment. rest looks good. Thanks @twthorn

[nssalian]

@twthorn do you have time to get to the pending comment to close this out?

[laskoviymishka]

LGTM!

[laskoviymishka]

Will wait @nssalian approve before merge.

[danielcweeks]

@twthorn can you please resolve all comments/conversations that have been addressed?

[twthorn]

@laskoviymishka @nssalian @danielcweeks Thank you all for the feedback. I have resolved & addressed all comments

[laskoviymishka]

LGTM!

[nssalian]

Thanks for addressing the comments. I'll defer to @danielcweeks

[danielcweeks]

@twthorn Minor comments. If you can address these, I think it's good to go.

[danielcweeks]

Minor: prefer full name topicPartition to tp

[danielcweeks]

@twthorn Just saw this other PR. It looks like it would also address the same issue, right: #17933

[vbhanuchander-lang]

@danielcweeks @twthorn Different fix, as far as I can tell — this one guards commitConsumerOffsets() against a stale coordinator committing over another's position, #17933 stops controlTopicOffsets itself from moving backwards in consumeAvailable() on a re-read. I ran each PR's tests against the other's fix and both sets fail, so neither subsumes the other; details and the test output are in #17933 (comment).

They touch different methods and do not conflict textually, so nothing here needs to wait on that one — I will rebase #17933 on top of this once it lands.

[nssalian]

@twthorn would be great to address the pending comments by Dan so we could get this in the release.

[twthorn]

@danielcweeks thanks for the review, updated with feedback!

[twthorn]

@danielcweeks These are different bugs dealing with different offset stores.

This PR addresses the issue of kafka consumer offsets being rewound (by stale coordinators). If rewound out of retention, the consumer falls back to its auto offset reset strategy, causing potential data loss.

The other PR addresses the offsets stored in the Iceberg snapshot summary (kafka.connect.offsets...). On restart, this value is used to skip records to avoid duplicates when committing to the iceberg table. When this value is rewound it leads to duplicate data.

So, separate bugs, both fixes needed. One to prevent data loss. The other to prevent duplicates.

[twthorn]

The only pending workflow Security Analysis with zizmor is stuck waiting for a runner Waiting for a runner to pick up this job... for about an hour. It normally completes in about 15 seconds, so this looks like a stall unrelated to the change. Requesting re-review now since we want to get this into the next release. If a maintainer can re-run the job it should pass.

Edit: a runner picked it up shortly after I posted this, all checks green

[danielcweeks]

Thanks @twthorn

[danielcweeks]

Thanks @twthorn !