You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "yadavay-amzn (via GitHub)" <gi...@apache.org> on 2026/05/07 06:28:59 UTC
[PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
yadavay-amzn opened a new pull request, #16237:
URL: https://github.com/apache/iceberg/pull/16237
Fixes #15878.
## Problem
The Kafka Connect `Coordinator` previously caught `Exception` around `doCommit()` and only logged a warning, so when a commit failed (e.g., a `CommitFailedException` from Glue detecting a concurrent table update), the connector stayed `RUNNING` while silently dropping the data that was in flight.
## Fix
Remove the catch-all around `doCommit()` and instead log at `ERROR` level with the task id and commit id before rethrowing. `CoordinatorThread.run()` already terminates the thread on uncaught exceptions, which transitions the Kafka Connect task to `FAILED` — so failures are now surfaced rather than dropped.
The `finally` block that calls `commitState.endCurrentCommit()` is preserved so per-commit state is cleaned up regardless of the outcome.
## Testing
- Added `testCommitFailedExceptionPropagates` which mocks a catalog-side `CommitFailedException` on `AppendFiles.commit()` and asserts it propagates out of `Coordinator.process()`. Without the fix, this test fails because the exception is swallowed.
- Updated two existing tests (`testCoordinatorWithBadDataFile` and `testCoordinatorCommittedOffsetValidation`) that previously relied on silent-swallow behaviour; they now assert the specific exception propagates (`IllegalArgumentException` for bad partition spec, `ValidationException` for stale offsets).
- Full `TestCoordinator` suite passes locally (8/8).
- `spotlessApply` passes.
## Generative AI
Generated-by: Claude Opus 4.7
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4410513674
Thanks @Baunsgaard for taking a look, you're right about the double-logging.
I've pushed an update with your recommended changes, please take a look when you get a chance. Thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #16237:
URL: https://github.com/apache/iceberg/pull/16237
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4466512844
@laskoviymishka @Baunsgaard Heads up — I made a change after your approvals that I want to flag:
The exception handling now uses a tiered approach instead of rethrowing everything:
1. `CommitFailedException` / `CommitStateUnknownException` → log + retry (transient conflicts)
2. Other `CleanableFailure` exceptions → log + retry (Iceberg's marker for recoverable errors)
3. Everything else → log ERROR + rethrow (fatal, kills the coordinator)
The reason: rethrowing ALL non-`CommitFailedException` exceptions was killing the coordinator on transient errors during the integration tests (the commit path can throw various `CleanableFailure` subtypes during normal catalog contention). The `CleanableFailure` interface is Iceberg's built-in classification for retryable exceptions, so using it as the boundary felt like the right approach.
CI is now green. Let me know if you'd prefer a different approach — e.g., a consecutive-failure counter that rethrows after N failures regardless of exception type.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3255032242
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -289,13 +313,14 @@ public void testCoordinatorCommittedOffsetValidation() {
Snapshot firstSnapshot = table.currentSnapshot();
assertThat(firstSnapshot.summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
- // Trigger commit to the table
- coordinatorTest(
- ImmutableList.of(EventTestUtil.createDataFile()), ImmutableList.of(), EventTestUtil.now());
-
- // Assert that the table was not updated and offsets remain
- table.refresh();
- assertThat(table.snapshots()).hasSize(2);
- assertThat(firstSnapshot.summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
+ // Trigger commit to the table - should throw ValidationException
+ assertThatThrownBy(
+ () ->
+ coordinatorTest(
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of(),
+ EventTestUtil.now()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("stale offsets");
Review Comment:
Same shape as the comment above, stale-offset conflict the table was *not* mutated (`snapshots().hasSize(2)` and offset still `{"0":7}`).
Those are the actual correctness guarantees of the optimistic-concurrency guard, and they're no longer checked. The exception-type assertion alone would pass even if a future regression committed the row delta before throwing.
Lets keep the post-throw assertions:
```java
assertThatThrownBy(...)
.isInstanceOf(ValidationException.class)
.hasMessageContaining("stale offsets");
table.refresh();
assertThat(table.snapshots()).hasSize(2);
assertThat(table.currentSnapshot().summary())
.containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
```
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -135,12 +141,30 @@ public void testCommitError() {
.withRecordCount(5)
.build();
- coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null);
-
- // no commit messages sent
- assertThat(producer.history()).hasSize(1);
+ assertThatThrownBy(
+ () -> coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot find partition spec");
Review Comment:
the prev-prev version of this test also asserted `producer.history().hasSize(1)` (only the `StartCommit` event was sent, no `CommitToTable`) and `table.snapshots().isEmpty()` (no phantom commit landed). v4 drops both.
The test now only proves an exception is thrown, it would still pass if a future regression sent a `CommitToTable` event before failing, or somehow committed-then-threw.
Maybe keep side-effect guards:
```java
assertThatThrownBy(...)
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Cannot find partition spec");
assertThat(producer.history()).hasSize(1);
assertThat(table.snapshots()).isEmpty();
```
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -150,12 +150,22 @@ protected boolean receive(Envelope envelope) {
private void commit(boolean partialCommit) {
try {
doCommit(partialCommit);
- } catch (Exception e) {
- LOG.warn(
- "Coordinator {} failed to commit for commit {}, will try again next cycle",
- taskId,
- commitState.currentCommitId(),
- e);
+ } catch (RuntimeException e) {
Review Comment:
Two small things on this catch block:
**SLF4J double-printing the message.** Both LOG calls pass `e.getMessage()` as a `{}` arg *and* `e` as the trailing throwable. SLF4J already prints the exception's message (and full stack) when the last arg is a `Throwable`, so the message ends up rendered twice in every log line.
**`taskId` asymmetry.** It's in the ERROR but dropped from the WARN. In a multi-task cluster you'd lose task identity for retried partial commits.
Both fixed at once:
```java
LOG.warn("Partial commit {} failed for task {}, will retry",
commitState.currentCommitId(), taskId, e);
...
LOG.error("Commit {} failed for task {}",
commitState.currentCommitId(), taskId, e);
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3216941834
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -148,13 +148,17 @@ protected boolean receive(Envelope envelope) {
}
private void commit(boolean partialCommit) {
+ // Do not swallow commit failures: wrap with Coordinator context and propagate so
+ // CoordinatorThread.run() terminates and the Kafka Connect task transitions to FAILED
+ // instead of silently dropping data (e.g., CommitFailedException from catalogs that
+ // detect concurrent updates). The taskId and commitId are embedded in the wrapper
+ // message so that the single log emitted by CoordinatorThread retains the context.
Review Comment:
nit: I think it is too much to leave this comment. It is a personal preference, but I would remove it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3253532652
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -150,12 +153,28 @@ protected boolean receive(Envelope envelope) {
private void commit(boolean partialCommit) {
try {
doCommit(partialCommit);
- } catch (Exception e) {
+ } catch (CommitFailedException | CommitStateUnknownException e) {
LOG.warn(
- "Coordinator {} failed to commit for commit {}, will try again next cycle",
- taskId,
+ "Commit {} failed, will retry on next cycle: {}",
commitState.currentCommitId(),
+ e.getMessage(),
e);
+ } catch (RuntimeException e) {
+ if (e instanceof CleanableFailure) {
Review Comment:
`CleanableFailure` doesn’t mean “safe to retry.”
Its only uncommitted metadata can be cleaned up.
This branch also catches things that are clearly not retryable:
* `ForbiddenException` / 403
* `BadRequestException` / 400
* `NotAuthorizedException` / 401
* `ValidationException`
So a bad credential could keep retrying forever at WARN, with no clear signal to the operator.
If we want retry behavior here, I’d make it explicit: list the retryable exception types instead of using `CleanableFailure` as the check.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -150,12 +153,28 @@ protected boolean receive(Envelope envelope) {
private void commit(boolean partialCommit) {
try {
doCommit(partialCommit);
- } catch (Exception e) {
+ } catch (CommitFailedException | CommitStateUnknownException e) {
Review Comment:
There are two different issues here, and both look bad.
`CommitStateUnknownException` should not be retried blindly. Its own Javadoc says we don’t know whether the commit succeeded, and retrying can create duplicates. Since the files stay in `commitBuffer`, the next cycle can commit the same files again. If the first commit actually landed, we get duplicate rows. This should be fatal and require manual table-state check.
`CommitFailedException` is also not something we should swallow. That is exactly the bug from #15878. A WARN + “retry next cycle” is not enough here: Kafka offsets may already be flushed, and on rebalance the in-memory commit state can be lost. Then the data is gone and the operator never sees a real failure.
Flink lets `CommitFailedException` propagate. I think we should do the same here.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -135,14 +141,50 @@ public void testCommitError() {
.withRecordCount(5)
.build();
- coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null);
+ assertThatThrownBy(
Review Comment:
`coordinatorTest()` calls `coordinator.process()` directly, so it skips the real production path.
In prod the flow is:
`CoordinatorThread.run()` catches the exception → marks `terminated = true` → next `CommitterImpl.save()` calls `processControlEvents()` → throws `NotRunningException`.
This test doesn’t cover that. It would still pass even if `CoordinatorThread` swallowed the failure.
I think we need an end-to-end test that goes through `CoordinatorThread` + `CommitterImpl`.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -135,14 +141,50 @@ public void testCommitError() {
.withRecordCount(5)
.build();
- coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null);
+ assertThatThrownBy(
+ () -> coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot find partition spec");
// no commit messages sent
assertThat(producer.history()).hasSize(1);
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ public void testCommitFailedExceptionSwallowed() {
Review Comment:
The comment says *“Verify issue #15878”*, but the test still expects `CommitFailedException` to be ignored.
That was the bug in #15878: the commit failure was swallowed and data was lost.
To verify the fix, the test should assert that the next `CommitterImpl.save()` throws `NotRunningException`, meaning the task moved to `FAILED` and the operator can see the failure.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4468228356
that's quite a swing @yadavay-amzn, and the new strategy changed enough that I need to flip to request some changes / questions.
The new behavior is basically:
> “If a commit fails with an Iceberg-shaped exception, WARN and hope the next cycle fixes it. Only fail on unexpected exceptions.”
I don’t think that’s safe. The previous direction was closer to correct:
> “If the real commit fails, kill the coordinator. If cleanup/partial commit fails, swallow it.”
Main issues:
1. `CommitStateUnknownException` is now swallowed. That one explicitly says retrying can create duplicates. Since the same files can still be retried from `commitBuffer`, this is exactly the bad case the exception warns about.
2. `CleanableFailure` is the wrong boundary. It means “metadata can be cleaned up,” not “safe to retry.” It also includes permanent errors like 400/401/403. With this change, bad credentials can loop forever at WARN, which is pretty close to the original bug.
3. `CommitFailedException` is still swallowed. That’s the exact exception from #15878. The “retry next cycle” story depends on a pretty fragile buffer/offset ordering assumption, and if a rebalance happens after offsets are committed, the in-memory state is gone and the failure is invisible.
I’d rather go back to the old shape: rethrow on full commit failure, swallow only the timed-out partial commit cleanup failure.
If we really want retries here, I think we need to explicitly list retryable exception types, bound the retries, and document the buffer lifecycle. Also the PR description drifted from the code again, so that needs an update too.
@AnatolyPopov — still interested in your read on the worker/coordinator offset ordering, since that’s the key assumption behind “retry next cycle.”
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4423332136
Done — removed the comment block.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4472315647
Addressed all three nits:
1. Fixed SLF4J double-printing -- removed `e.getMessage()` from format args, added `taskId` to WARN for symmetry.
2. `testCommitError` -- added `producer.history().hasSize(1)` and `table.snapshots().isEmpty()` assertions.
3. `testCoordinatorCommittedOffsetValidation` -- added post-throw assertions that table still has 2 snapshots and offsets unchanged.
Thanks @laskoviymishka!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4469504269
@laskoviymishka You are right -- the `CleanableFailure` approach was too permissive. Reverted to the simpler and safer design:
- **Full commit failure**: rethrow everything (kills coordinator -> task FAILED)
- **Partial commit failure** (timed-out cleanup): WARN + swallow (retry next cycle)
This matches your original direction. The `CleanableFailure` detour was me trying to work around a CI integration test failure that turned out to be transient -- the test passes 10/10 locally with this approach.
PR description updated to match the 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3208678316
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -148,14 +148,18 @@ protected boolean receive(Envelope envelope) {
}
private void commit(boolean partialCommit) {
+ // Do not swallow commit failures: propagate them so CoordinatorThread terminates
+ // and the Kafka Connect task transitions to FAILED instead of silently dropping data
+ // (e.g., CommitFailedException from catalogs that detect concurrent updates).
try {
doCommit(partialCommit);
- } catch (Exception e) {
- LOG.warn(
- "Coordinator {} failed to commit for commit {}, will try again next cycle",
+ } catch (RuntimeException e) {
+ LOG.error(
+ "Coordinator {} failed to commit for commit {}; propagating failure to terminate task",
taskId,
commitState.currentCommitId(),
e);
+ throw e;
Review Comment:
change it to
```
throw new RuntimeException(
String.format("Coordinator %s failed to commit %s",
taskId, commitState.currentCommitId()),
e);
```
This allows the further up `CoordinatorThread.run()` catch to log the error once, and still attribute the error to this location.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -135,14 +141,40 @@ public void testCommitError() {
.withRecordCount(5)
.build();
- coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null);
+ assertThatThrownBy(
+ () -> coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot find partition spec");
// no commit messages sent
assertThat(producer.history()).hasSize(1);
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ public void testCommitFailedExceptionPropagates() {
+ // Reproduce issue #15878: a CommitFailedException from the catalog (e.g., Glue concurrent
+ // update) must propagate out of Coordinator.process() so that CoordinatorThread terminates
+ // and the Kafka Connect task transitions to FAILED instead of silently dropping data.
+ Table spiedTable = spy(table);
+ AppendFiles spiedAppend = spy(table.newAppend());
+ doThrow(new CommitFailedException("Glue detected concurrent update"))
+ .when(spiedAppend)
+ .commit();
+ when(spiedTable.newAppend()).thenReturn(spiedAppend);
+ when(catalog.loadTable(TABLE_IDENTIFIER)).thenReturn(spiedTable);
+
+ assertThatThrownBy(
+ () ->
+ coordinatorTest(
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of(),
+ EventTestUtil.now()))
+ .isInstanceOf(CommitFailedException.class)
+ .hasMessageContaining("Glue detected concurrent update");
Review Comment:
if you do the above change, then i think this need to be :
```
.hasRootCauseMessage("Glue detected concurrent update");
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4469495394
@laskoviymishka You're right -- the `CleanableFailure` approach was too permissive. Reverted to the simpler and safer design:
- **Full commit failure**: rethrow everything (kills coordinator -> task FAILED)
- **Partial commit failure** (timed-out cleanup): WARN + swallow (retry next cycle)
This matches your original direction. The `CleanableFailure` detour was me trying to work around a CI integration test failure that turned out to be transient -- the test passes 10/10 locally with this approach.
PR description updated to match the 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3211813885
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -148,14 +148,18 @@ protected boolean receive(Envelope envelope) {
}
private void commit(boolean partialCommit) {
+ // Do not swallow commit failures: propagate them so CoordinatorThread terminates
+ // and the Kafka Connect task transitions to FAILED instead of silently dropping data
+ // (e.g., CommitFailedException from catalogs that detect concurrent updates).
try {
doCommit(partialCommit);
- } catch (Exception e) {
- LOG.warn(
- "Coordinator {} failed to commit for commit {}, will try again next cycle",
+ } catch (RuntimeException e) {
+ LOG.error(
+ "Coordinator {} failed to commit for commit {}; propagating failure to terminate task",
taskId,
commitState.currentCommitId(),
e);
+ throw e;
Review Comment:
Done, updated in latest revision
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on code in PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#discussion_r3211819064
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -135,14 +141,40 @@ public void testCommitError() {
.withRecordCount(5)
.build();
- coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null);
+ assertThatThrownBy(
+ () -> coordinatorTest(ImmutableList.of(badDataFile), ImmutableList.of(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot find partition spec");
// no commit messages sent
assertThat(producer.history()).hasSize(1);
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ public void testCommitFailedExceptionPropagates() {
+ // Reproduce issue #15878: a CommitFailedException from the catalog (e.g., Glue concurrent
+ // update) must propagate out of Coordinator.process() so that CoordinatorThread terminates
+ // and the Kafka Connect task transitions to FAILED instead of silently dropping data.
+ Table spiedTable = spy(table);
+ AppendFiles spiedAppend = spy(table.newAppend());
+ doThrow(new CommitFailedException("Glue detected concurrent update"))
+ .when(spiedAppend)
+ .commit();
+ when(spiedTable.newAppend()).thenReturn(spiedAppend);
+ when(catalog.loadTable(TABLE_IDENTIFIER)).thenReturn(spiedTable);
+
+ assertThatThrownBy(
+ () ->
+ coordinatorTest(
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of(),
+ EventTestUtil.now()))
+ .isInstanceOf(CommitFailedException.class)
+ .hasMessageContaining("Glue detected concurrent update");
Review Comment:
Done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4462034549
Thanks for the thorough review @laskoviymishka. Addressed all points:
1. **Rethrow original exception** -- no longer wrapping in `RuntimeException`. The original `CommitFailedException` (or whatever the commit throws) is logged at ERROR and rethrown directly, preserving the type for alerting.
2. **Partial-commit gating** -- rethrow is now gated on `!partialCommit`. Transient failures during partial commits are logged but not fatal, matching the previous retry behavior. Only full-commit failures terminate the coordinator.
3. **Updated test assertions** -- tests now expect the original exception types directly.
Will update the PR description to accurately reflect the narrowed catch and the mechanism for task FAILED transition.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4469484817
@laskoviymishka You're right — the `CleanableFailure` approach was too permissive. Reverted to the simpler and safer design:
- **Full commit failure**: rethrow everything (kills coordinator → task FAILED)
- **Partial commit failure** (timed-out cleanup): WARN + swallow (retry next cycle)
This matches your original direction. The `CleanableFailure` detour was me trying to work around a CI integration test failure that turned out to be transient — the test passes 10/10 locally with this approach.
PR description updated to match the 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4469514463
@laskoviymishka You are right -- the `CleanableFailure` approach was too permissive. Reverted to the simpler and safer design:
- **Full commit failure**: rethrow everything (kills coordinator -> task FAILED)
- **Partial commit failure** (timed-out cleanup): WARN + swallow (retry next cycle)
This matches your original direction. CI is green now. PR description updated to match the code.
On the test coverage point (coordinatorTest skipping the production CoordinatorThread path) -- agreed that an end-to-end test through CoordinatorThread + CommitterImpl would be stronger. Happy to add that in this PR or as a follow-up, whichever you prefer.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Surface commit failures instead of silently swallowing them [iceberg]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #16237:
URL: https://github.com/apache/iceberg/pull/16237#issuecomment-4479961483
Thanks @AnatolyPopov for the thorough review and the context on the offset ordering! And thanks @laskoviymishka for driving this to a good shape.
Agreed on the follow-ups -- I will create issues to track:
1. Metrics for partial commit failures (operator visibility)
2. Bounded retry on transient exceptions (avoid operator intervention for transitive errors)
Both are strictly additive and can land independently.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org