diff --git a/docs/generated/iceberg_configuration.html b/docs/generated/iceberg_configuration.html index c08388c8cbc4..435227216d31 100644 --- a/docs/generated/iceberg_configuration.html +++ b/docs/generated/iceberg_configuration.html @@ -122,6 +122,18 @@ String Metastore table name for Iceberg Catalog.Set this as an iceberg table alias if using a centralized Catalog. + +
metadata.iceberg.unknown-host-retry.initial-delay-ms
+ 1000 + Long + Initial delay in milliseconds before retrying an Iceberg REST catalog DNS lookup failure. The delay doubles after each failure. + + +
metadata.iceberg.unknown-host-retry.max-retries
+ 5 + Integer + Maximum number of retries after an Iceberg REST catalog DNS lookup failure. Set to 0 to disable retries. +
metadata.iceberg.uri
(none) diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java index 819865066d12..113c413d710f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java @@ -95,6 +95,20 @@ public class IcebergOptions { "The number of old metadata files to keep after each table commit. " + "For rest-catalog, it will keep 1 old metadata at least."); + public static final ConfigOption UNKNOWN_HOST_RETRY_MAX_RETRIES = + key("metadata.iceberg.unknown-host-retry.max-retries") + .intType() + .defaultValue(5) + .withDescription( + "Maximum number of retries after an Iceberg REST catalog DNS lookup failure. Set to 0 to disable retries."); + + public static final ConfigOption UNKNOWN_HOST_RETRY_INITIAL_DELAY_MILLIS = + key("metadata.iceberg.unknown-host-retry.initial-delay-ms") + .longType() + .defaultValue(1_000L) + .withDescription( + "Initial delay in milliseconds before retrying an Iceberg REST catalog DNS lookup failure. The delay doubles after each failure."); + public static final ConfigOption URI = key("metadata.iceberg.uri") .stringType() 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..3377bec5bdf3 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 @@ -57,6 +57,7 @@ import java.io.IOException; import java.lang.reflect.Field; +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -90,12 +91,25 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { private final String icebergDatabaseName; private final TableIdentifier icebergTableIdentifier; private final IcebergOptions icebergOptions; + private final int unknownHostMaxRetries; + private final long unknownHostInitialRetryDelayMillis; private Table icebergTable; public IcebergRestMetadataCommitter(FileStoreTable table) { Options options = new Options(table.options()); icebergOptions = new IcebergOptions(options); + unknownHostMaxRetries = options.get(IcebergOptions.UNKNOWN_HOST_RETRY_MAX_RETRIES); + unknownHostInitialRetryDelayMillis = + options.get(IcebergOptions.UNKNOWN_HOST_RETRY_INITIAL_DELAY_MILLIS); + Preconditions.checkArgument( + unknownHostMaxRetries >= 0, + "%s must be non-negative", + IcebergOptions.UNKNOWN_HOST_RETRY_MAX_RETRIES.key()); + Preconditions.checkArgument( + unknownHostInitialRetryDelayMillis >= 0, + "%s must be non-negative", + IcebergOptions.UNKNOWN_HOST_RETRY_INITIAL_DELAY_MILLIS.key()); this.fileIO = table.fileIO(); this.metadataDirectory = IcebergCommitCallback.catalogTableMetadataPath(table); @@ -140,15 +154,56 @@ 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); + long delayMillis = unknownHostInitialRetryDelayMillis; + for (int retry = 0; ; retry++) { + try { + commitMetadataImpl(newIcebergMetadata, baseIcebergMetadata); + return; + } catch (Exception e) { + if (!hasUnknownHostCause(e) || retry == unknownHostMaxRetries) { + throw commitFailure(e); + } + + LOG.warn( + "Iceberg REST catalog DNS lookup failed for table {}; retrying in {} ms " + + "({}/{}).", + icebergTableIdentifier, + delayMillis, + retry + 1, + unknownHostMaxRetries, + e); + try { + sleepBeforeUnknownHostRetry(delayMillis); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw commitFailure(interrupted); + } + delayMillis = delayMillis > Long.MAX_VALUE / 2 ? Long.MAX_VALUE : delayMillis * 2; + } } } - private void commitMetadataImpl( + private RuntimeException commitFailure(Exception cause) { + return new RuntimeException( + "Fail to commit iceberg metadata for table: " + icebergTableIdentifier, cause); + } + + @VisibleForTesting + protected void sleepBeforeUnknownHostRetry(long delayMillis) throws InterruptedException { + Thread.sleep(delayMillis); + } + + private static boolean hasUnknownHostCause(Throwable failure) { + for (Throwable cause = failure; cause != null; cause = cause.getCause()) { + if (cause instanceof UnknownHostException) { + return true; + } + } + return false; + } + + @VisibleForTesting + protected void commitMetadataImpl( IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) { newIcebergMetadata = adjustMetadataForRest(newIcebergMetadata); 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..bb24cad57d2b 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 @@ -70,6 +70,9 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1730,6 +1733,140 @@ public void testRegisteredMetadataFilesAreWriteOnce() throws Exception { } } + @Test + public void testUnknownHostRetriesWithExponentialBackoff() throws Exception { + Map options = new HashMap<>(); + options.put(IcebergOptions.UNKNOWN_HOST_RETRY_MAX_RETRIES.key(), "2"); + options.put(IcebergOptions.UNKNOWN_HOST_RETRY_INITIAL_DELAY_MILLIS.key(), "7"); + FileStoreTable table = createRetryTestTable(options); + IcebergMetadata metadata = writeLocalMetadata(table); + TestingIcebergRestMetadataCommitter committer = + new TestingIcebergRestMetadataCommitter( + table, + 2, + new RuntimeException(new UnknownHostException("simulated DNS failure"))); + + committer.commitMetadata(metadata, null); + + assertThat(committer.attempts).isEqualTo(3); + assertThat(committer.delays).containsExactly(7L, 14L); + assertThat( + restCatalog + .loadTable(TableIdentifier.of("mydb", "t")) + .currentSnapshot() + .snapshotId()) + .isEqualTo(1); + } + + @Test + public void testUnknownHostRetryExhaustionFailsCommit() throws Exception { + Map options = new HashMap<>(); + options.put(IcebergOptions.UNKNOWN_HOST_RETRY_MAX_RETRIES.key(), "1"); + options.put(IcebergOptions.UNKNOWN_HOST_RETRY_INITIAL_DELAY_MILLIS.key(), "7"); + FileStoreTable table = createRetryTestTable(options); + IcebergMetadata metadata = writeLocalMetadata(table); + TestingIcebergRestMetadataCommitter committer = + new TestingIcebergRestMetadataCommitter( + table, + Integer.MAX_VALUE, + new RuntimeException(new UnknownHostException("simulated DNS failure"))); + + assertThatThrownBy(() -> committer.commitMetadata(metadata, null)) + .hasRootCauseInstanceOf(UnknownHostException.class); + assertThat(committer.attempts).isEqualTo(2); + assertThat(committer.delays).containsExactly(7L); + assertThat(restCatalog.tableExists(TableIdentifier.of("mydb", "t"))).isFalse(); + } + + @Test + public void testZeroUnknownHostRetriesPreservesImmediateFailure() throws Exception { + Map options = new HashMap<>(); + options.put(IcebergOptions.UNKNOWN_HOST_RETRY_MAX_RETRIES.key(), "0"); + FileStoreTable table = createRetryTestTable(options); + IcebergMetadata metadata = writeLocalMetadata(table); + TestingIcebergRestMetadataCommitter committer = + new TestingIcebergRestMetadataCommitter( + table, + Integer.MAX_VALUE, + new RuntimeException(new UnknownHostException("simulated DNS failure"))); + + assertThatThrownBy(() -> committer.commitMetadata(metadata, null)) + .hasRootCauseInstanceOf(UnknownHostException.class); + assertThat(committer.attempts).isEqualTo(1); + assertThat(committer.delays).isEmpty(); + } + + @Test + public void testNonDnsFailureIsNotRetried() throws Exception { + FileStoreTable table = createRetryTestTable(Collections.emptyMap()); + IcebergMetadata metadata = writeLocalMetadata(table); + TestingIcebergRestMetadataCommitter committer = + new TestingIcebergRestMetadataCommitter( + table, Integer.MAX_VALUE, new IllegalStateException("catalog failure")); + + assertThatThrownBy(() -> committer.commitMetadata(metadata, null)) + .hasRootCauseInstanceOf(IllegalStateException.class); + assertThat(committer.attempts).isEqualTo(1); + assertThat(committer.delays).isEmpty(); + } + + private FileStoreTable createRetryTestTable(Map options) throws Exception { + restCatalog.dropTable(TableIdentifier.of("mydb", "t"), false); + return createPaimonTable( + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}), + Collections.emptyList(), + Collections.emptyList(), + -1, + "avro", + options); + } + + private static IcebergMetadata writeLocalMetadata(FileStoreTable table) throws Exception { + FileStoreTable localTable = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location")); + String commitUser = UUID.randomUUID().toString(); + try (TableWriteImpl write = localTable.newWrite(commitUser); + TableCommitImpl commit = localTable.newCommit(commitUser)) { + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(true, 1)); + } + return localMetadata(localTable, 1); + } + + private static class TestingIcebergRestMetadataCommitter extends IcebergRestMetadataCommitter { + + private int failuresRemaining; + private final RuntimeException failure; + private int attempts; + private final List delays = new ArrayList<>(); + + private TestingIcebergRestMetadataCommitter( + FileStoreTable table, int failures, RuntimeException failure) { + super(table); + this.failuresRemaining = failures; + this.failure = failure; + } + + @Override + protected void commitMetadataImpl( + IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) { + attempts++; + if (failuresRemaining > 0) { + failuresRemaining--; + throw failure; + } + super.commitMetadataImpl(newIcebergMetadata, baseIcebergMetadata); + } + + @Override + protected void sleepBeforeUnknownHostRetry(long delayMillis) { + delays.add(delayMillis); + } + } + private static IcebergMetadata localMetadata(FileStoreTable table, long snapshotId) { return IcebergMetadata.fromPath( table.fileIO(),