From 00dc939a4233535fdc6ffc826782d57c45b1fd21 Mon Sep 17 00:00:00 2001 From: arthurgaubil Date: Fri, 11 Sep 2026 17:05:58 -0400 Subject: [PATCH] [iceberg] Do not fail the commit when the Iceberg REST catalog rejects or is unsure IcebergRestMetadataCommitter calls TableOperations.commit() directly and lets every exception escape. Because the Iceberg sync runs inside a Paimon commit callback, which for Flink runs inside notifyCheckpointComplete, any exception there is fatal: the whole job restarts. For a job syncing many tables, one table losing a commit race stops all of them. Two of these exceptions do not warrant that. CommitStateUnknownException means the outcome is unknown, and CommitFailedException means the compare-and-swap was rejected and nothing was applied (it implements CleanableFailure). In both cases the next commit attempt reloads the table and runs checkBase() against the live catalog state, which either matches and proceeds or detects the drift and rebuilds from the current file set. Paimon's own commit has already durably applied the data, so only the Iceberg metadata lags, by one commit. Log a warning and let the next attempt reconcile instead of failing the job. Any other exception still propagates unchanged. Closes #8875 --- .../iceberg/IcebergRestMetadataCommitter.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java index 1115624788dd..70860ff26bef 100644 --- a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java +++ b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java @@ -46,6 +46,8 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.rest.Endpoint; import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.types.Types; @@ -254,6 +256,37 @@ private void commitMetadataImpl( ((BaseTable) icebergTable) .operations() .commit(((BaseTable) icebergTable).operations().current(), updatedForCommit); + } catch (CommitStateUnknownException e) { + // The catalog returned an ambiguous response, so we cannot tell whether this commit + // was applied server-side. Either way the next attempt reloads the table and runs + // checkBase() against that live state: if it landed, the base matches and the next + // commit proceeds normally; if it did not, checkBase() sees the drift and the table + // is rebuilt from the current file set. Failing here does not resolve the ambiguity, + // it only takes down every other table the job is syncing. + LOG.warn( + "Commit to rest catalog returned an ambiguous response for table {}, snapshot" + + " {}; not failing the commit, the next attempt will reconcile.", + icebergTableIdentifier, + updatedForCommit.currentSnapshot() == null + ? null + : updatedForCommit.currentSnapshot().snapshotId(), + e); + } catch (CommitFailedException e) { + // The catalog rejected the compare-and-swap because the table moved between our read + // of the base metadata and this commit. Unlike the ambiguous case above this one is + // unambiguous: CommitFailedException implements CleanableFailure, so nothing landed + // server-side. It reconciles through the same path on the next attempt, and Paimon's + // own commit has already durably applied the write, so only the Iceberg metadata + // lags, by one commit. + LOG.warn( + "Commit to rest catalog was rejected for table {}, snapshot {}, because the" + + " table changed concurrently; not failing the commit, the next" + + " attempt will reconcile.", + icebergTableIdentifier, + updatedForCommit.currentSnapshot() == null + ? null + : updatedForCommit.currentSnapshot().snapshotId(), + e); } catch (Exception e) { throw new RuntimeException( "Fail to commit metadata to rest catalog for table: " + icebergTableIdentifier,