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..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 @@ -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; @@ -80,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"; @@ -140,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); + } } } @@ -251,9 +270,10 @@ private void commitMetadataImpl( } try { - ((BaseTable) icebergTable) - .operations() - .commit(((BaseTable) icebergTable).operations().current(), updatedForCommit); + 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, @@ -261,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 {