Kafka Connect: Track control topic offsets as a high-water mark

[vbhanuchander-lang]

Closes #17340.

Opening this at @ericyangliu's invitation on the issue — the diagnosis and the production evidence
(149 double-referenced files, ~112k duplicated rows) are his.

The bug

Channel.consumeAvailable recorded the consumed position with an unconditional put:

controlTopicOffsets.put(record.partition(), record.offset() + 1);

Nothing compares against the value already stored, so any re-read of a control topic partition —
a rebalance resuming the consumer from the last committed group offsets, as in the report — moves
the tracked position backwards.

That regression is durable rather than transient, because the map is not just bookkeeping:

• commitConsumerOffsets() commits it for the consumer group, so the next restart resumes from the
  regressed offset and re-reads more.
• Coordinator.commitToTable merges it into kafka.connect.offsets on the snapshot, which is the
  watermark the min-offset filter uses on subsequent commits.

Once the watermark is behind, replayed DataWritten envelopes pass that filter. distinctByKey
only dedupes within one commit and append does no path-level dedup, so the same data files are
committed again and every scan reads them twice.

The change

One line: keep the highest position seen for the partition.

controlTopicOffsets.merge(record.partition(), record.offset() + 1, Long::max);

This is what every reader of controlTopicOffsets() already assumes — commitToTable even folds
it in with Long::max against the last committed offsets. Making the map itself monotonic is
consistent with that, and it does not change the offsets recorded on the forward path.

Tests

TestChannel drives a Channel over a MockConsumer. Consuming offsets 0-4 reaches a watermark
of 5; a seek back to 1 then delivers a partial replay ending at offset 2, which is the shape of the
re-read in the report.

• controlTopicOffsetsTrackTheHighestPositionConsumed — the map stays at 5. With the fix reverted
  it is 3.
• committedControlTopicOffsetsDoNotRegressOnReplay — asserts what the channel actually commits to
  Kafka, OffsetAndMetadata{offset=5}. With the fix reverted it commits 3, which is the offset a
  restarted channel would resume from.
• controlTopicOffsetsAreTrackedPerPartition — partitions stay independent.

Both regression tests fail on main without the change. Full module suite passes (136 tests), as
do spotlessCheck and checkstyle.

Not included

The issue also floats a bounded set of recently committed file locations in the coordinator as a
content-level backstop for replays the offset arithmetic cannot see. That is a design call for
maintainers and a larger change, so I have left it out rather than hold up the correctness fix.
Happy to follow up if it is wanted.

[uros-b]

Looks like a correct fix, so thank you @vbhanuchander-lang! Should definitely ping relevant committers for further review, e.g. @bryanck who created the IKC sink and commit-coordination path or @danielcweeks

[vbhanuchander-lang]

Thanks @uros-b — and you pointed at the right person: #10351 ("Kafka Connect: Commit coordination")
is where consumeAvailable and controlTopicOffsets came from.

@bryanck @danielcweeks a short summary so this is quick to judge, since it is a one-line change with
a longer argument behind it.

Channel.consumeAvailable recorded the consumed position with an unconditional
put(partition, offset + 1), so re-reading a control topic partition moves the tracked position
backwards. The map is not just local bookkeeping — it feeds commitConsumerOffsets(), which is
the group offset a restarted channel resumes from, and it is merged into kafka.connect.offsets on
the snapshot, which is the watermark the min-offset filter uses on later commits. Once that watermark
is behind, replayed DataWritten envelopes pass the filter, distinctByKey only dedupes within one
commit, and append does no path-level dedup — so the same data files are committed twice. The
reporter of #17340 saw 149 double-referenced files and ~112k duplicated rows in production.

The argument I would most like checked: commitToTable already folds controlTopicOffsets()
with Long::max against the last committed offsets, so its only consumers already treat this value
as "furthest position consumed". Making the map itself monotonic just brings it in line with that,
and it does not change the offsets recorded on the forward path.

TestChannel is new — there was no direct Channel test before. It drives a real Channel over a
MockConsumer: consume offsets 0-4 to reach a watermark of 5, then seek back to 1 and deliver a
partial replay ending at 2, which is the shape of the re-read in the report. One test asserts the
in-memory map, one asserts what the channel actually commits to Kafka. Both fail on main at 3
instead of 5. The partial tail matters — replaying the whole 0-4 range leaves put at 5 and passes
with the bug present.

One open question for you rather than for the diff: #17340 also floats a bounded set of recently
committed file locations in the coordinator, as a content-level backstop for replays the offset
arithmetic cannot see. I left it out deliberately — it is a design call and a larger change, and I did
not want it holding up the correctness fix. If you want it, I am happy to follow up separately.

[danielcweeks]

@twthorn Does this address the same issue as #17552?

[vbhanuchander-lang]

Answered in the timeline — short version: different method, different cause, and each PR's tests fail against the other's fix.

[vbhanuchander-lang]

@danielcweeks @twthorn They overlap in symptom but not in cause, and neither one subsumes the other — I checked by running each PR's tests against the other's fix.

#17552 guards the write: a per-instance committedOffsets cache in commitConsumerOffsets() that refuses to commit behind what this instance last committed. That is the zombie-coordinator case in #17551 — coordinator A's tracked position is legitimately its own, and the rewind comes from A committing over B. This PR guards the value: controlTopicOffsets itself moves backwards in consumeAvailable() when a partition is re-read, so it is already wrong before any commit path sees it.

Applying #17552's Channel.java onto main, keeping main's put in consumeAvailable, and running this PR's TestChannel:

committedControlTopicOffsetsDoNotRegressOnReplay FAILED   expected offset=5, was offset=3
controlTopicOffsetsTrackTheHighestPositionConsumed FAILED expected {0=5}, was {0=3}

#17552's guard does not fire here because the cache is empty on an instance's first commit — lastCommittedOffset == null commits unconditionally — which is exactly the state a coordinator is in right after the rebalance that caused the re-read.

The reverse also holds. This PR's Channel.java on main with #17552's TestCoordinator:

testCommitConsumerOffsetsDoesNotRewind FAILED             expected 100, was 5
testCommitConsumerDuplicateDoesNotCommit FAILED           expected 105, was 100
testCommitConsumerMixedPartitionsRewindOrAdvance FAILED   expected 200, was 195

Making one instance's map monotonic says nothing about another coordinator's position.

One further gap: doCommit passes controlTopicOffsets() to commitToTable before commitConsumerOffsets() runs, and that value is stamped into kafka.connect.offsets.*. The Long::max fold there protects a table that already has an offset for the partition, but a table with no prior entry for it gets the rewound value written — #17552's guard sits downstream of that write and cannot reach it.

The two are complementary and do not conflict textually — different methods in the same file. #17552 is further along, so I am happy to rebase this on top of it once it lands.