From 192548bcdd56ebabd6de8949374baa92485415b9 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Fri, 21 Aug 2026 18:59:54 +0000 Subject: [PATCH] improve IT infra --- .../beam/it/gcp/dataflow/AbstractPipelineLauncher.java | 4 ++-- .../cloud/teleport/it/iceberg/IcebergResourceManager.java | 8 ++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java index 68677d8422..1c0cb7ec8c 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java @@ -316,9 +316,9 @@ protected final LaunchInfo getJobInfo(LaunchConfig options, JobState state, Job public JobState waitUntilActive(String project, String region, String jobId) throws IOException { JobState state = getJobStatus(project, region, jobId); boolean logOnce = false; - while (PENDING_STATES.contains(state)) { + while (PENDING_STATES.contains(state) || state == JobState.UNKNOWN) { if (!logOnce) { - LOG.info("Job still pending. Will check again in 15 seconds"); + LOG.info("Job still pending or unknown. Will check again in 15 seconds"); logOnce = true; } try { diff --git a/it/iceberg/src/main/java/com/google/cloud/teleport/it/iceberg/IcebergResourceManager.java b/it/iceberg/src/main/java/com/google/cloud/teleport/it/iceberg/IcebergResourceManager.java index 53a99b6b10..cbbabed302 100644 --- a/it/iceberg/src/main/java/com/google/cloud/teleport/it/iceberg/IcebergResourceManager.java +++ b/it/iceberg/src/main/java/com/google/cloud/teleport/it/iceberg/IcebergResourceManager.java @@ -300,6 +300,14 @@ public synchronized void cleanupAll() throws IcebergResourceManagerException { dropNamespace(namespace, true); } createdNamespaces.clear(); + if (cachedCatalog instanceof AutoCloseable) { + try { + ((AutoCloseable) cachedCatalog).close(); + } catch (Exception e) { + LOG.warn("Error closing Iceberg catalog", e); + } + } + cachedCatalog = null; LOG.info("Cleaned up all resources for test ID: {}.", testId); }