From 2e0216c500820ea958dce97115a1992dabae7569 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 12 Nov 2026 06:08:06 +0000 Subject: [PATCH 1/2] Retry catalogCleanup on ForbiddenException in BigQueryMetastoreCatalogIT --- .../catalog/BigQueryMetastoreCatalogIT.java | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java index a0c412596014..a08580382a2c 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java @@ -38,7 +38,9 @@ import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.ForbiddenException; import org.junit.Test; public class BigQueryMetastoreCatalogIT extends IcebergCatalogBaseIT { @@ -89,6 +91,26 @@ public Map managedIcebergConfig(String tableId) { .build(); } + @Override + public void catalogCleanup(List namespaces) throws IOException { + for (int attempt = 1; attempt <= 3; attempt++) { + try { + super.catalogCleanup(namespaces); + return; + } catch (ForbiddenException e) { + if (attempt == 3) { + throw e; + } + try { + Thread.sleep(5000L * attempt); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting to retry BigQuery catalog cleanup", ie); + } + } + } + } + @Test public void testWriteToPartitionedAndValidateWithBQQuery() throws IOException, InterruptedException { From 7039747db859ae246f684932735501bb37e0190d Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 13 Nov 2026 05:09:05 +0000 Subject: [PATCH 2/2] Move catalogCleanup ForbiddenException retry into IcebergCatalogBaseIT --- .../catalog/BigQueryMetastoreCatalogIT.java | 22 ---------------- .../iceberg/catalog/IcebergCatalogBaseIT.java | 26 +++++++++++++++---- 2 files changed, 21 insertions(+), 27 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java index a08580382a2c..a0c412596014 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java @@ -38,9 +38,7 @@ import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.catalog.Catalog; -import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.exceptions.ForbiddenException; import org.junit.Test; public class BigQueryMetastoreCatalogIT extends IcebergCatalogBaseIT { @@ -91,26 +89,6 @@ public Map managedIcebergConfig(String tableId) { .build(); } - @Override - public void catalogCleanup(List namespaces) throws IOException { - for (int attempt = 1; attempt <= 3; attempt++) { - try { - super.catalogCleanup(namespaces); - return; - } catch (ForbiddenException e) { - if (attempt == 3) { - throw e; - } - try { - Thread.sleep(5000L * attempt); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - throw new IOException("Interrupted while waiting to retry BigQuery catalog cleanup", ie); - } - } - } - } - @Test public void testWriteToPartitionedAndValidateWithBQQuery() throws IOException, InterruptedException { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index 95ad34af13e1..c7ba912a4845 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -113,6 +113,7 @@ import org.apache.iceberg.deletes.PositionDeleteWriter; import org.apache.iceberg.encryption.EncryptedFiles; import org.apache.iceberg.encryption.InputFilesDecryptor; +import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.DataWriter; import org.apache.iceberg.io.FileIO; @@ -188,11 +189,26 @@ public void catalogSetup() { public void catalogCleanup(List namespaces) throws IOException { for (Namespace namespace : namespaces) { - for (TableIdentifier identifier : catalog.listTables(namespace)) { - catalog.dropTable(identifier); - } - if (catalog instanceof SupportsNamespaces) { - ((SupportsNamespaces) catalog).dropNamespace(namespace); + for (int attempt = 1; attempt <= 3; attempt++) { + try { + for (TableIdentifier identifier : catalog.listTables(namespace)) { + catalog.dropTable(identifier); + } + if (catalog instanceof SupportsNamespaces) { + ((SupportsNamespaces) catalog).dropNamespace(namespace); + } + break; + } catch (ForbiddenException e) { + if (attempt == 3) { + throw e; + } + try { + Thread.sleep(SETUP_TEARDOWN_SLEEP_MS * attempt); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting to retry catalog cleanup", ie); + } + } } } }