Skip to content

[iceberg] Retry REST metadata publication before cleanup #9749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arthurgaubil wants to merge 2 commits into
base: master
Choose a base branch
from
+169 −8
Open

[iceberg] Retry REST metadata publication before cleanup #9749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
00dc939
[iceberg] Do not fail the commit when the Iceberg REST catalog reject…
arthurgaubil Sep 11, 2026
969f70f
[iceberg] Reconcile REST publication before cleanup
arthurgaubil Sep 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 33 additions & 8 deletions paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -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);
}
}
}

Expand Down Expand Up @@ -251,16 +270,22 @@ 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,
e);
}
}

@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);
Expand Down
136 changes: 136 additions & 0 deletions paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<String, String> 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(),
Expand All @@ -1749,6 +1830,61 @@ private static List<Path> 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 {
Expand Down
Loading