Add iceberg.jdbc-catalog.retryable-status-codes configuration property

Allows retrying additional, JDBC driver and database specific errors such as Postgres status codes `57000,57P03,57P04` if using Postgres.
trinodb · Nov 16, 2024 · 8e25e5c · 8e25e5c
1 parent 495e5c0
commit 8e25e5c
Show file tree

Hide file tree

Showing 5 changed files with 28 additions and 0 deletions.
diff --git a/docs/src/main/sphinx/object-storage/metastores.md b/docs/src/main/sphinx/object-storage/metastores.md
@@ -563,6 +563,14 @@ directory.
 * - `iceberg.jdbc-catalog.schema-version`
   - JDBC catalog schema version.
     Valid values are `V0` or `V1`. Defaults to `V1`.
+* - `iceberg.jdbc-catalog.retryable-status-codes`
+  - On connection error to JDBC metastore, retry if
+    it is one of these JDBC status codes.
+    Valid value is a comma-separated list of status codes.
+    Note: JDBC catalog always retries the following status
+    codes: `08000,08003,08006,08007,40001`. Specify only
+    additional codes (such as `57000,57P03,57P04` if using
+    PostgreSQL driver) here.
 :::
 
 :::{warning}

diff --git a/...-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/IcebergJdbcCatalogConfig.java b/...-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/IcebergJdbcCatalogConfig.java
@@ -35,6 +35,7 @@ public enum SchemaVersion
     private String catalogName;
     private String defaultWarehouseDir;
     private SchemaVersion schemaVersion = SchemaVersion.V1;
+    private String retryableStatusCodes;
 
     @NotNull
     public String getDriverClass()
@@ -134,4 +135,18 @@ public IcebergJdbcCatalogConfig setSchemaVersion(SchemaVersion schemaVersion)
         this.schemaVersion = schemaVersion;
         return this;
     }
+
+    @NotNull
+    public Optional<String> getRetryableStatusCodes()
+    {
+        return Optional.ofNullable(retryableStatusCodes);
+    }
+
+    @Config("iceberg.jdbc-catalog.retryable-status-codes")
+    @ConfigDescription("On connection error to JDBC metastore, retry if it is one of these JDBC status codes")
+    public IcebergJdbcCatalogConfig setRetryableStatusCodes(String retryableStatusCodes)
+    {
+        this.retryableStatusCodes = retryableStatusCodes;
+        return this;
+    }
 }
diff --git a/...o-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/TrinoJdbcCatalogFactory.java b/...o-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/TrinoJdbcCatalogFactory.java
@@ -74,6 +74,7 @@ public TrinoJdbcCatalogFactory(
         properties.put(PROPERTY_PREFIX + "schema-version", jdbcConfig.getSchemaVersion().toString());
         jdbcConfig.getConnectionUser().ifPresent(user -> properties.put(PROPERTY_PREFIX + "user", user));
         jdbcConfig.getConnectionPassword().ifPresent(password -> properties.put(PROPERTY_PREFIX + "password", password));
+        jdbcConfig.getRetryableStatusCodes().ifPresent(codes -> properties.put("retryable_status_codes", codes));
         this.catalogProperties = properties.buildOrThrow();
 
         this.clientPool = new JdbcClientPool(jdbcConfig.getConnectionUrl(), catalogProperties);

diff --git a/...berg/src/test/java/io/trino/plugin/iceberg/catalog/jdbc/TestIcebergJdbcCatalogConfig.java b/...berg/src/test/java/io/trino/plugin/iceberg/catalog/jdbc/TestIcebergJdbcCatalogConfig.java
@@ -35,6 +35,7 @@ public void testDefaults()
                 .setConnectionPassword(null)
                 .setCatalogName(null)
                 .setDefaultWarehouseDir(null)
+                .setRetryableStatusCodes(null)
                 .setSchemaVersion(SchemaVersion.V1));
     }
 
@@ -48,6 +49,7 @@ public void testExplicitPropertyMappings()
                 .put("iceberg.jdbc-catalog.connection-password", "bar")
                 .put("iceberg.jdbc-catalog.catalog-name", "test")
                 .put("iceberg.jdbc-catalog.default-warehouse-dir", "s3://bucket")
+                .put("iceberg.jdbc-catalog.retryable-status-codes", "57P01,57P05")
                 .put("iceberg.jdbc-catalog.schema-version", "V0")
                 .buildOrThrow();
 
@@ -58,6 +60,7 @@ public void testExplicitPropertyMappings()
                 .setConnectionPassword("bar")
                 .setCatalogName("test")
                 .setDefaultWarehouseDir("s3://bucket")
+                .setRetryableStatusCodes("57P01,57P05")
                 .setSchemaVersion(SchemaVersion.V0);
 
         assertFullMapping(properties, expected);

diff --git a/...t/java/io/trino/plugin/iceberg/catalog/jdbc/TestIcebergJdbcCatalogConnectorSmokeTest.java b/...t/java/io/trino/plugin/iceberg/catalog/jdbc/TestIcebergJdbcCatalogConnectorSmokeTest.java
@@ -109,6 +109,7 @@ protected QueryRunner createQueryRunner()
                                 .put("iceberg.register-table-procedure.enabled", "true")
                                 .put("iceberg.writer-sort-buffer-size", "1MB")
                                 .put("iceberg.jdbc-catalog.default-warehouse-dir", warehouseLocation.getAbsolutePath())
+                                .put("iceberg.jdbc-catalog.retryable-status-codes", "57P01,57P05")
                                 .buildOrThrow())
                 .setInitialTables(REQUIRED_TPCH_TABLES)
                 .build();