Core: Make metadata refresh retries configurable in JDBC catalog

core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java

@@ -93,6 +93,7 @@ public class JdbcCatalog extends BaseMetastoreViewCatalog
   private JdbcClientPool connections;
   private Map<String, String> catalogProperties;
   private boolean uniqueTableLocation;
+  private int metadataRefreshMaxRetries;
   private final Function<Map<String, String>, FileIO> ioBuilder;
   private final Function<Map<String, String>, JdbcClientPool> clientPoolBuilder;
   private boolean initializeCatalogTables;
@@ -130,6 +131,7 @@ public void initialize(String name, Map<String, String> properties) {
             properties,
             CatalogProperties.UNIQUE_TABLE_LOCATION,
             CatalogProperties.UNIQUE_TABLE_LOCATION_DEFAULT);
+    this.metadataRefreshMaxRetries = JdbcUtil.metadataRefreshMaxRetries(properties);
 
     if (name != null) {
       this.catalogName = name;
@@ -284,15 +286,22 @@ private static boolean executeV1CatalogUpdate(Connection conn) throws SQLExcepti
   @Override
   protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
     return new JdbcTableOperations(
-        connections, io, catalogName, tableIdentifier, catalogProperties, schemaVersion);
+        connections,
+        io,
+        catalogName,
+        tableIdentifier,
+        catalogProperties,
+        schemaVersion,
+        metadataRefreshMaxRetries);
   }
 
   @Override
   protected ViewOperations newViewOps(TableIdentifier viewIdentifier) {
     if (schemaVersion != JdbcUtil.SchemaVersion.V1) {
       throw new UnsupportedOperationException(VIEW_WARNING_LOG_MESSAGE);
     }
-    return new JdbcViewOperations(connections, io, catalogName, viewIdentifier, catalogProperties);
+    return new JdbcViewOperations(
+        connections, io, catalogName, viewIdentifier, catalogProperties, metadataRefreshMaxRetries);
   }
 
   @Override

core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java

@@ -49,20 +49,23 @@ class JdbcTableOperations extends BaseMetastoreTableOperations {
   private final JdbcClientPool connections;
   private final Map<String, String> catalogProperties;
   private final JdbcUtil.SchemaVersion schemaVersion;
+  private final int metadataRefreshMaxRetries;
 
   protected JdbcTableOperations(
       JdbcClientPool dbConnPool,
       FileIO fileIO,
       String catalogName,
       TableIdentifier tableIdentifier,
       Map<String, String> catalogProperties,
-      JdbcUtil.SchemaVersion schemaVersion) {
+      JdbcUtil.SchemaVersion schemaVersion,
+      int metadataRefreshMaxRetries) {
     this.catalogName = catalogName;
     this.tableIdentifier = tableIdentifier;
     this.fileIO = fileIO;
     this.connections = dbConnPool;
     this.catalogProperties = catalogProperties;
     this.schemaVersion = schemaVersion;
+    this.metadataRefreshMaxRetries = metadataRefreshMaxRetries;
   }
 
   @Override
@@ -96,7 +99,7 @@ public void doRefresh() {
         newMetadataLocation != null,
         "Invalid table %s: metadata location is null",
         tableIdentifier);
-    refreshFromMetadataLocation(newMetadataLocation);
+    refreshFromMetadataLocation(newMetadataLocation, metadataRefreshMaxRetries);
   }
 
   @Override

core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java

@@ -44,6 +44,10 @@ final class JdbcUtil {
   // property to control if catalog tables are created during initialization
   static final String INIT_CATALOG_TABLES_PROPERTY =
       JdbcCatalog.PROPERTY_PREFIX + "init-catalog-tables";
+  // property to control how many times reading table or view metadata is retried during a refresh
+  static final String METADATA_REFRESH_MAX_RETRIES_PROPERTY =
+      JdbcCatalog.PROPERTY_PREFIX + "metadata-refresh-max-retries";
+  static final int METADATA_REFRESH_MAX_RETRIES_DEFAULT = 20;
 
   static final String RETRYABLE_STATUS_CODES = "retryable_status_codes";
   private static final String POSTGRES_UNIQUE_VIOLATION_SQLSTATE = "23505";
@@ -528,6 +532,20 @@ static Properties filterAndRemovePrefix(Map<String, String> properties, String p
     return result;
   }
 
+  static int metadataRefreshMaxRetries(Map<String, String> properties) {
+    int maxRetries =
+        PropertyUtil.propertyAsInt(
+            properties,
+            METADATA_REFRESH_MAX_RETRIES_PROPERTY,
+            METADATA_REFRESH_MAX_RETRIES_DEFAULT);
+    Preconditions.checkArgument(
+        maxRetries >= 0,
+        "Invalid %s: %s (must be >= 0)",
+        METADATA_REFRESH_MAX_RETRIES_PROPERTY,
+        maxRetries);
+    return maxRetries;
+  }
+
   static boolean isConstraintViolation(SQLException ex) {
     return ex instanceof SQLIntegrityConstraintViolationException
         || POSTGRES_UNIQUE_VIOLATION_SQLSTATE.equals(ex.getSQLState())

core/src/main/java/org/apache/iceberg/jdbc/JdbcViewOperations.java

@@ -49,18 +49,36 @@ public class JdbcViewOperations extends BaseViewOperations {
   private final FileIO fileIO;
   private final JdbcClientPool connections;
   private final Map<String, String> catalogProperties;
+  private final int metadataRefreshMaxRetries;
 
   protected JdbcViewOperations(
       JdbcClientPool dbConnPool,
       FileIO fileIO,
       String catalogName,
       TableIdentifier viewIdentifier,
       Map<String, String> catalogProperties) {
+    this(
+        dbConnPool,
+        fileIO,
+        catalogName,
+        viewIdentifier,
+        catalogProperties,
+        JdbcUtil.metadataRefreshMaxRetries(catalogProperties));
+  }
+
+  JdbcViewOperations(
+      JdbcClientPool dbConnPool,
+      FileIO fileIO,
+      String catalogName,
+      TableIdentifier viewIdentifier,
+      Map<String, String> catalogProperties,
+      int metadataRefreshMaxRetries) {
     this.catalogName = catalogName;
     this.viewIdentifier = viewIdentifier;
     this.fileIO = fileIO;
     this.connections = dbConnPool;
     this.catalogProperties = catalogProperties;
+    this.metadataRefreshMaxRetries = metadataRefreshMaxRetries;
   }
 
   @Override
@@ -90,7 +108,7 @@ protected void doRefresh() {
     String newMetadataLocation = view.get(JdbcTableOperations.METADATA_LOCATION_PROP);
     Preconditions.checkState(
         newMetadataLocation != null, "Invalid view %s: metadata location is null", viewIdentifier);
-    refreshFromMetadataLocation(newMetadataLocation);
+    refreshFromMetadataLocation(newMetadataLocation, null, metadataRefreshMaxRetries);
   }
 
   @Override

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java

@@ -24,9 +24,12 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -54,6 +57,7 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
@@ -65,6 +69,7 @@
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.hadoop.Util;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -1067,6 +1072,41 @@ public void testRemoveProperties() {
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
   }
 
+  @Test
+  public void testMetadataRefreshMaxRetries() {
+    try (JdbcCatalog jdbcCatalog =
+        initCatalog(
+            "test_jdbc_catalog_refresh_retries",
+            ImmutableMap.of(JdbcUtil.METADATA_REFRESH_MAX_RETRIES_PROPERTY, "1"))) {
+      TableIdentifier tableIdent = TableIdentifier.of("db", "tbl");
+      jdbcCatalog.buildTable(tableIdent, SCHEMA).create();
+
+      try (MockedStatic<TableMetadataParser> mockedStatic =
+          Mockito.mockStatic(TableMetadataParser.class)) {
+        mockedStatic
+            .when(() -> TableMetadataParser.read(any(FileIO.class), anyString()))
+            .thenThrow(new UncheckedIOException(new IOException("Read timed out")));
+
+        assertThatThrownBy(() -> jdbcCatalog.loadTable(tableIdent))
+            .isInstanceOf(UncheckedIOException.class)
+            .hasMessageContaining("Read timed out");
+        mockedStatic.verify(
+            () -> TableMetadataParser.read(any(FileIO.class), anyString()), times(2));
+      }
+    }
+  }
+
+  @Test
+  public void testInvalidMetadataRefreshMaxRetries() {
+    assertThatThrownBy(
+            () ->
+                initCatalog(
+                    "test_jdbc_catalog_invalid_refresh_retries",
+                    ImmutableMap.of(JdbcUtil.METADATA_REFRESH_MAX_RETRIES_PROPERTY, "-1")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid jdbc.metadata-refresh-max-retries: -1 (must be >= 0)");
+  }
+
   @Test
   public void testConversions() {
     Namespace ns = Namespace.of("db", "db2", "ns2");

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcViewCatalog.java

@@ -21,8 +21,12 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
 
 import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.sql.SQLException;
 import java.util.Map;
 import java.util.UUID;
@@ -33,10 +37,13 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.view.BaseView;
 import org.apache.iceberg.view.ViewCatalogTests;
 import org.apache.iceberg.view.ViewMetadata;
+import org.apache.iceberg.view.ViewMetadataParser;
 import org.apache.iceberg.view.ViewOperations;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -52,6 +59,10 @@ public class TestJdbcViewCatalog extends ViewCatalogTests<JdbcCatalog> {
 
   @BeforeEach
   public void before() {
+    catalog = initCatalog(ImmutableMap.of());
+  }
+
+  private JdbcCatalog initCatalog(Map<String, String> additionalProperties) {
     Map<String, String> properties = Maps.newHashMap();
     properties.put(
         CatalogProperties.URI,
@@ -65,10 +76,12 @@ public void before() {
     properties.put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key3", "catalog-default-key3");
     properties.put(CatalogProperties.VIEW_OVERRIDE_PREFIX + "key3", "catalog-override-key3");
     properties.put(CatalogProperties.VIEW_OVERRIDE_PREFIX + "key4", "catalog-override-key4");
+    properties.putAll(additionalProperties);
 
-    catalog = new JdbcCatalog();
-    catalog.setConf(new Configuration());
-    catalog.initialize("testCatalog", properties);
+    JdbcCatalog jdbcCatalog = new JdbcCatalog();
+    jdbcCatalog.setConf(new Configuration());
+    jdbcCatalog.initialize("testCatalog", properties);
+    return jdbcCatalog;
   }
 
   @Override
@@ -91,6 +104,34 @@ protected boolean supportsEmptyNamespace() {
     return true;
   }
 
+  @Test
+  public void testMetadataRefreshMaxRetries() {
+    try (JdbcCatalog jdbcCatalog =
+        initCatalog(ImmutableMap.of(JdbcUtil.METADATA_REFRESH_MAX_RETRIES_PROPERTY, "0"))) {
+      TableIdentifier identifier = TableIdentifier.of("namespace1", "view");
+      jdbcCatalog.createNamespace(identifier.namespace());
+      jdbcCatalog
+          .buildView(identifier)
+          .withQuery("spark", "select * from tbl")
+          .withSchema(SCHEMA)
+          .withDefaultNamespace(identifier.namespace())
+          .create();
+
+      try (MockedStatic<ViewMetadataParser> mockedStatic =
+          Mockito.mockStatic(ViewMetadataParser.class)) {
+        mockedStatic
+            .when(() -> ViewMetadataParser.read(any(FileIO.class), anyString()))
+            .thenThrow(new UncheckedIOException(new IOException("Read timed out")));
+
+        assertThatThrownBy(() -> jdbcCatalog.loadView(identifier))
+            .isInstanceOf(UncheckedIOException.class)
+            .hasMessageContaining("Read timed out");
+        mockedStatic.verify(
+            () -> ViewMetadataParser.read(any(FileIO.class), anyString()), times(1));
+      }
+    }
+  }
+
   @Test
   public void testCommitExceptionWithoutMessage() {
     TableIdentifier identifier = TableIdentifier.of("namespace1", "view");

docs/docs/jdbc.md

@@ -35,6 +35,7 @@ the JDBC catalog allows arbitrary configurations through:
 | -------------------- | --------------------------------- | ------------------------------------------------------ |
 | uri                  |                                   | the JDBC connection string |
 | jdbc.<property_key\> |                                   | any key value pairs to configure the JDBC connection |
+| jdbc.metadata-refresh-max-retries | 20                      | number of times to retry reading table or view metadata during a refresh |
 
 ### Examples