From 00dc939a4233535fdc6ffc826782d57c45b1fd21 Mon Sep 17 00:00:00 2001 From: arthurgaubil Date: Fri, 11 Sep 2026 17:05:58 -0400 Subject: [PATCH 1/2] [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, From 969f70fcd16aa40b857d297e7c38ee892f62dd08 Mon Sep 17 00:00:00 2001 From: Arthur Gaubil Date: Tue, 15 Sep 2026 17:01:10 -0400 Subject: [PATCH 2/2] [iceberg] Reconcile REST publication before cleanup --- .../iceberg/IcebergRestMetadataCommitter.java | 70 ++++----- .../IcebergRestMetadataCommitterTest.java | 136 ++++++++++++++++++ 2 files changed, 167 insertions(+), 39 deletions(-) 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 70860ff26bef..9d050554b305 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 @@ -82,6 +82,8 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { private static final String PAIMON_COMMIT_IDENTITY = "paimon-commit-identity"; + private static final int MAX_COMMIT_ATTEMPTS = 3; + private static final Logger LOG = LoggerFactory.getLogger(IcebergRestMetadataCommitter.class); private static final String REST_CATALOG_NAME = "rest-catalog"; @@ -142,11 +144,26 @@ public void commitMetadata(Path newMetadataPath, @Nullable Path baseMetadataPath @Override public void commitMetadata( IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) { - try { - commitMetadataImpl(newIcebergMetadata, baseIcebergMetadata); - } catch (Exception e) { - throw new RuntimeException( - "Fail to commit iceberg metadata for table: " + icebergTableIdentifier, e); + for (int attempt = 1; attempt <= MAX_COMMIT_ATTEMPTS; attempt++) { + try { + commitMetadataImpl(newIcebergMetadata, baseIcebergMetadata); + return; + } catch (CommitStateUnknownException | CommitFailedException e) { + if (attempt == MAX_COMMIT_ATTEMPTS) { + throw new RuntimeException( + "Fail to commit iceberg metadata for table: " + icebergTableIdentifier, + e); + } + LOG.warn( + "Commit attempt {} to rest catalog failed for table {}; reloading catalog" + + " state before retrying.", + attempt, + icebergTableIdentifier, + e); + } catch (Exception e) { + throw new RuntimeException( + "Fail to commit iceberg metadata for table: " + icebergTableIdentifier, e); + } } } @@ -253,40 +270,10 @@ private void commitMetadataImpl( } try { - ((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); + BaseTable table = (BaseTable) icebergTable; + commit(table, table.operations().current(), updatedForCommit); + } catch (CommitStateUnknownException | CommitFailedException e) { + throw e; } catch (Exception e) { throw new RuntimeException( "Fail to commit metadata to rest catalog for table: " + icebergTableIdentifier, @@ -294,6 +281,11 @@ private void commitMetadataImpl( } } + @VisibleForTesting + protected void commit(BaseTable table, TableMetadata base, TableMetadata updated) { + table.operations().commit(base, updated); + } + private TableMetadata.Builder updatesForCorrectBase( TableMetadata base, TableMetadata newMetadata, boolean isNewTable) { TableMetadata.Builder updateBuilder = TableMetadata.buildFrom(base); diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java index b0a1afcae512..39db3730fa69 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java @@ -56,6 +56,8 @@ import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.data.IcebergGenerics; import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.hadoop.HadoopCatalog; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -1730,6 +1732,85 @@ public void testRegisteredMetadataFilesAreWriteOnce() throws Exception { } } + @Test + public void testPublicationFailuresReconcileBeforeSuccess() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map options = new HashMap<>(); + options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "2"); + options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "2"); + FileStoreTable catalogTable = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + "avro", + options); + FileStoreTable disabledTable = + catalogTable.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "disabled")); + + String commitUser = UUID.randomUUID().toString(); + FailingIcebergRestMetadataCommitter committer = + new FailingIcebergRestMetadataCommitter(catalogTable); + IcebergCommitCallback callback = new IcebergCommitCallback(catalogTable, commitUser); + setMetadataCommitter(callback, committer); + TableWriteImpl write = disabledTable.newWrite(commitUser); + TableCommitImpl commit = disabledTable.newCommit(commitUser); + TableIdentifier identifier = TableIdentifier.of("mydb", "t"); + + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(true, 1)); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(1)); + + committer.failNext(FailureType.COMMIT_FAILED, false, 1); + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(true, 2)); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(2)); + assertThat(committer.commitCalls()).isEqualTo(2); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(2); + + committer.failNext(FailureType.STATE_UNKNOWN, true, 1); + write.write(GenericRow.of(3, 30)); + commit.commit(3, write.prepareCommit(true, 3)); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(3)); + assertThat(committer.commitCalls()).isEqualTo(1); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3); + + committer.failNext(FailureType.STATE_UNKNOWN, false, 1); + write.write(GenericRow.of(4, 40)); + commit.commit(4, write.prepareCommit(true, 4)); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(4)); + assertThat(committer.commitCalls()).isEqualTo(2); + assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(4); + + committer.failNext(FailureType.COMMIT_FAILED, false, 10); + write.write(GenericRow.of(5, 50)); + commit.commit(5, write.prepareCommit(true, 5)); + assertThatThrownBy( + () -> callback.retry(new org.apache.paimon.manifest.ManifestCommittable(5))) + .hasRootCauseInstanceOf(CommitFailedException.class); + assertThat(committer.commitCalls()).isEqualTo(3); + write.close(); + commit.close(); + callback.close(); + + Table published = restCatalog.loadTable(identifier); + assertThat(published.currentSnapshot().snapshotId()).isEqualTo(4); + for (org.apache.iceberg.Snapshot snapshot : published.snapshots()) { + assertThat(catalogTable.fileIO().exists(new Path(snapshot.manifestListLocation()))) + .isTrue(); + assertThat(snapshot.allManifests(published.io())).isNotEmpty(); + } + IcebergPathFactory pathFactory = + new IcebergPathFactory( + IcebergCommitCallback.catalogTableMetadataPath(catalogTable)); + assertThat(catalogTable.fileIO().exists(pathFactory.toMetadataPath(3))).isTrue(); + } + private static IcebergMetadata localMetadata(FileStoreTable table, long snapshotId) { return IcebergMetadata.fromPath( table.fileIO(), @@ -1749,6 +1830,61 @@ private static List registerFiles(FileStoreTable table) throws Exception { return files; } + private static void setMetadataCommitter( + IcebergCommitCallback callback, IcebergMetadataCommitter committer) throws Exception { + java.lang.reflect.Field field = + IcebergCommitCallback.class.getDeclaredField("metadataCommitter"); + field.setAccessible(true); + field.set(callback, committer); + } + + private enum FailureType { + COMMIT_FAILED, + STATE_UNKNOWN + } + + private static class FailingIcebergRestMetadataCommitter extends IcebergRestMetadataCommitter { + + private FailureType failureType; + private boolean applyBeforeThrow; + private int failuresRemaining; + private int commitCalls; + + private FailingIcebergRestMetadataCommitter(FileStoreTable table) { + super(table); + } + + private void failNext(FailureType failureType, boolean applyBeforeThrow, int failures) { + this.failureType = failureType; + this.applyBeforeThrow = applyBeforeThrow; + this.failuresRemaining = failures; + this.commitCalls = 0; + } + + private int commitCalls() { + return commitCalls; + } + + @Override + protected void commit(BaseTable table, TableMetadata base, TableMetadata updated) { + commitCalls++; + if (failuresRemaining == 0) { + super.commit(table, base, updated); + return; + } + + failuresRemaining--; + if (applyBeforeThrow) { + super.commit(table, base, updated); + } + if (failureType == FailureType.COMMIT_FAILED) { + throw new CommitFailedException("injected rejected commit"); + } + throw new CommitStateUnknownException( + "injected ambiguous commit", new RuntimeException("injected")); + } + } + /** Makes the committer's REST client see a server that does not advertise registerTable. */ private static void removeRegisterTableEndpoint(IcebergRestMetadataCommitter committer) throws Exception {