kumarpritam863 commented on PR #16084:
URL: https://github.com/apache/iceberg/pull/16084#issuecomment-5099876918
@laskoviymishka thanks for such and extensive review and fruitful and
important suggestions. I have made the changes as per you suggestions:
1. stop() always completes cleanup (Worker.java:334)
terminateBackgroundPolling() now never throws — it logs at ERROR on
timeout, and on interrupt it re-asserts the. interrupt flag after cleanup so it
can't abort the producer/writer close. Cleanup runs in a finally.
- The polling thread now owns the consumer's whole lifecycle including
close(). Previously a hung poller plus a consumer.close() from the task thread
would hit ConcurrentModificationException; now the second close from stop() is
just a no-op.
- stop() is idempotent, and stopWorker()/stopCoordinator() null their
field in a finally.
2. Startup race closed (Worker.java:160)
start() blocks on a latch counted down after subscribe + initial poll, and
surfaces a subscription failure instead of returning as if it succeeded. This
matters more than the review implied: the worker's control group is transient
with auto.offset.reset=latest, so the position reset itself is the cutoff —
deferring subscribe was widening the window, not just delaying detection.
3. Recovery model, made explicit (Channel.java:135, CommitterImpl.java:209)
Two halves:
- A single undecodable control record is now skipped, counted, and logged
at ERROR rather than killing the poller. The offset was already advanced, so
skipping is the only option that makes progress; missing a START_COMMIT this
way only defers the task's data to the next round.
- A genuinely fatal poller failure fails the task permanently.
CommitterImpl records a sticky failure and refuses further records, and I
removed the stopWorker() from the error path.
That's the fix for the silent-data-loss case: no replacement worker can
answer the outstanding START_COMMIT with an empty result. Restart from
committed offsets is now the documented recovery path.
4. Tests — 11 in TestWorker (was 6), all with try/finally worker.stop().
The pollDelay(200ms) sleep is replaced with consumedRecordCount(), so "consumed
and ignored" is distinguishablefrom "not read yet";
>= 2 is now hasSize(2); added second-process()-after-failure (asserted
3×), poison-pill survival, subscription-failure, stop-idempotency, and
cleanup-when-sinkWriter.close()-throws.
@danielcweeks on this `Are these observed issues or just potential`:
This is due to both. We observed that intermittently the workers miss the
start_commit due to Duration.Zero which sometime cause a record to be available
in the Iceberg Table after a delay of twice or sometime thrice the
commit-interval.
--
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]
