nahidupa opened a new pull request, #18006:
URL: https://github.com/apache/iceberg/pull/18006
## Problem
The coordinator's control-topic consumer group receives a committed offset
in exactly one place — `Coordinator.doCommit`, after every table has committed
successfully:
```java
// we should only get here if all tables committed successfully...
commitConsumerOffsets();
commitState.clearResponses();
```
Until that first success the `<group>-coord` group has **no committed
offset**, so `auto.offset.reset` applies. `KafkaClientFactory` defaults it to
`latest`:
```java
consumerProps.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
```
A coordinator holds its `DataWritten` responses in memory
(`CommitState.commitBuffer`). If it is replaced before that first successful
commit — any rebalance that moves the leader partition will do it — the buffer
is discarded with the object, and `Channel.stop()` does not commit offsets. The
replacement starts at the **log end** and never re-reads the responses its
predecessor consumed.
Those files are never registered in the table. They are not redelivered
either, because `Channel.send()` publishes `DataWritten` and commits the
worker's source offsets in a single transaction:
```java
recordList.forEach(producer::send);
if (!sourceOffsets.isEmpty()) {
producer.sendOffsetsToTransaction(offsetsToCommit,
KafkaUtils.consumerGroupMetadata(context));
}
producer.commitTransaction();
```
Once that transaction commits, the source records are consumed for good. The
only record that the files exist is the control-topic message the replacement
just skipped.
The result is silent: no exception, no warning, and the files sit
unreferenced in object storage. The window is widest at connector startup,
which is exactly when Connect rebalances most, and it **reopens** whenever the
group's offsets expire (`offsets.retention.minutes`, default 7 days) during an
idle period.
## Fix
Pass the reset strategy through `KafkaClientFactory` and `Channel`, and have
the coordinator ask for `earliest`.
Re-reading is safe by design. The authoritative dedup floor is the per-table
`kafka.connect.offsets.<topic>.<group>` property in the snapshot summary, and
`commitToTable` already filters against it:
```java
Long minOffset = committedOffsets.get(envelope.partition());
return minOffset == null || envelope.offset() >= minOffset;
```
Anything genuinely committed is filtered out; anything buffered but
uncommitted is recovered. That is what the floor is for.
The worker deliberately keeps `latest`. Its group is transient
(`controlGroupIdPrefix() + UUID.randomUUID()`), is never committed to, and only
needs `StartCommit` messages published after it starts — it must not replay
control-topic history on every restart.
## Tests
`TestCoordinatorOffsetReset`:
- `coordinatorConsumerReadsFromEarliest` — the durable group asks for
`earliest`.
- `workerConsumerReadsFromLatest` — the transient group still asks for
`latest`.
- `replacementCoordinatorRecoversFilesBufferedByItsPredecessor` —
end-to-end: a file is announced to one coordinator, that coordinator is
replaced without ever committing an offset, and the file must still reach the
table.
The end-to-end test builds each `MockConsumer` from the strategy the
production code actually requested, so it exercises the real wiring rather than
a hard-coded strategy.
Verified to fail without the fix:
```
coordinatorConsumerReadsFromEarliest() FAILED
replacementCoordinatorRecoversFilesBufferedByItsPredecessor() FAILED
3 tests completed, 2 failed
```
and to pass with it, along with the existing suite and `spotlessCheck`.
`ChannelTestBase` gains a stub for the new `createConsumer` overload.
## Why the existing suite never caught this
`ChannelTestBase` constructs its `MockConsumer` with
`OffsetResetStrategy.EARLIEST`, while production defaults to `latest`. The
harness could not express the failure. A probe confirms `MockConsumer` does
honour the strategy:
```
PROBE strategy=earliest position=1 recordsReturned=1
PROBE strategy=latest position=1 recordsReturned=0
```
## Scope and risk
Deliberately narrow — one behaviour, three source files.
The main risk is startup cost: a brand-new connect group against a
long-lived control topic will now read that topic from the beginning once.
Events for other groups are skipped by the existing
`event.groupId().equals(connectGroupId)` check, and same-group events below the
floor are filtered, so the work is bounded and happens once per group.
`putIfAbsent` still lets operators override via
`iceberg.kafka.auto.offset.reset`.
Related but **not** addressed here: #16282's rebalance-replay mechanism (see
#17713, #17933), and coordinator election hardening (#17376, #17450). This is
an independent gap that none of them close — it is caused by coordinator
*restart*, not by replay.
---
**AI Disclosure**
- Model: Claude Opus 4.6
- Platform/Tool: opencode
- Human Oversight: partially reviewed
- Prompt Summary: An AI agent audited the Kafka Connect sink coordinator,
identified this gap, wrote the reproduction, implemented the fix, and drafted
this description. An independent model review found fixture flaws in an earlier
version of the reproduction; those were corrected and the reproduction was
re-verified to fail before the fix and pass after. Code and description are
AI-generated and partially reviewed by the author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
