jun-su-lee opened a new pull request, #17593:
URL: https://github.com/apache/iceberg/pull/17593
## Problem
A sink task that holds partitions but receives no records never answers the
coordinator's commit requests. The coordinator waits for responses covering
every
partition in the consumer group, so one such task stalls every commit round:
the
round times out after `iceberg.control.commit.timeout-ms` and falls back to a
partial commit. Data still lands, but commit latency is pinned to the timeout
instead of completing as soon as the writers report, and partial commits
never
set the `kafka.connect.valid-through-ts` snapshot property, so the watermark
stops advancing.
## Root cause
The worker is this task's side of the commit protocol. Besides writing
records,
it reports every assigned partition on a commit request, filling in a null
offset
for partitions that received no data. `Worker#receive` states the intent
plainly:
```java
// include all assigned topic partitions even if no messages were read
// from a partition, as the coordinator will use that to determine
// when all data for a commit has been received
List<TopicPartitionOffset> assignments =
context.assignment().stream()
.map(
tp -> {
Offset offset = results.sourceOffsets().get(tp);
if (offset == null) {
offset = Offset.NULL_OFFSET;
}
...
```
That code is correct. What was wrong is that nothing reached it: the worker's
lifetime was tied to records rather than to the assignment.
`CommitterImpl#save`
created it only inside the branch guarded by
`sinkRecords != null && !sinkRecords.isEmpty()`, so an idle task never had a
worker for the block above to run in.
`close` stops the worker on every rebalance to discard uncommitted work, and
a
cooperative rebalance that only revokes partitions is not followed by
`open`, so
a task can keep partitions while its worker stays gone until the next record.
## Fix
Start the worker whenever the task holds an assignment:
```java
if (ownsPartitions()) {
startWorker();
if (sinkRecords != null && !sinkRecords.isEmpty()) {
worker.save(sinkRecords);
}
}
processControlEvents();
```
`ownsPartitions()` reads `SinkTaskContext#assignment`, so it follows the
framework's own view instead of tracking partitions separately. The
`isInitialized` check keeps a task that was never assigned any partitions --
and
therefore never saw `open` -- from dereferencing a context it does not have.
`save` is the only place this can be done: it is where `processControlEvents`
polls the control topic, so a worker started anywhere else would exist
without
ever being driven.
## Testing
`TestCommitterImplWorkerLifecycle` adds three cases:
| Test | Covers |
| --- | --- |
| `idleTaskAnswersCommitRequestForAllAssignedPartitions` | a task that never
receives a record still reports every assigned partition, with null offsets |
| `idleTaskAnswersCommitRequestAfterRebalance` | a cooperative rebalance
that revokes one partition leaves the task able to answer for the rest, with a
freshly created worker |
| `fullyRevokedTaskDoesNotRestartWorker` | a task that lost every partition
does not get a worker back |
Verified by mutation: reverting `CommitterImpl`, making `Worker#process` a
no-op,
removing `stopWorker` from `close`, or dropping the assignment check each
turns
the suite red. The `isInitialized` check is pinned by the existing
`TestCommitterImpl` cases, which call `save` on a committer that never saw
`open`.
Full module suite: 133 tests, 0 failures. `spotlessCheck` passes.
## Assumptions reviewers may want to check
The fix rests on two Kafka Connect behaviors that were confirmed by reading
the
AK 4.3 source rather than by running a cluster:
- `WorkerSinkTask.HandleRebalance#onPartitionsAssigned` returns early on an
empty
collection, so a rebalance that only revokes partitions does not call
`open`.
This is why the worker has to be restarted from `save`.
- `WorkerSinkTask#deliverMessages` calls `put` unconditionally, so an idle
task
still receives an empty collection on every poll iteration. The `SinkTask`
contract does not promise this.
The new tests mock the framework, so they pin `CommitterImpl`'s behavior
under
those assumptions without verifying the assumptions themselves.
## Known limitation, unchanged by this PR
A worker subscribes to the control topic with a fresh group ID and
`auto.offset.reset=latest`, so a worker created after the coordinator has
already
published a `StartCommit` misses that round. This patch narrows the window to
just after a rebalance rather than the whole idle period, but does not close
it.
---
**AI Disclosure**
- Model: Claude Opus 5
- Platform/Tool: Claude Code
- Human Oversight: fully reviewed
- Prompt Summary: Investigate why idle Iceberg sink tasks stop answering
commit
requests after a rebalance, then fix the worker lifecycle so it follows the
partition assignment rather than the arrival of records.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
