[iceberg] Retry transient REST DNS failures

docs/generated/iceberg_configuration.html

@@ -122,6 +122,18 @@
             <td>String</td>
             <td>Metastore table name for Iceberg Catalog.Set this as an iceberg table alias if using a centralized Catalog.</td>
         </tr>
+        <tr>
+            <td><h5>metadata.iceberg.unknown-host-retry.initial-delay-ms</h5></td>
+            <td style="word-wrap: break-word;">1000</td>
+            <td>Long</td>
+            <td>Initial delay in milliseconds before retrying an Iceberg REST catalog DNS lookup failure. The delay doubles after each failure.</td>
+        </tr>
+        <tr>
+            <td><h5>metadata.iceberg.unknown-host-retry.max-retries</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>Maximum number of retries after an Iceberg REST catalog DNS lookup failure. Set to 0 to disable retries.</td>
+        </tr>
         <tr>
             <td><h5>metadata.iceberg.uri</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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<Integer> 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<Long> 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<String> URI =
             key("metadata.iceberg.uri")
                     .stringType()

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);

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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<Long> 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(),