From 50b1044a10653bd115a7baa57eea80871b802946 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Tue, 18 Aug 2026 15:28:47 -0400 Subject: [PATCH 1/4] Fix DataStream test timeout Datastream finalizes and rotates GCS files every 60 seconds, triggering GCS OBJECT_FINALIZE Pub/Sub notifications promptly so the integration test receives data and completes normally without hitting the condition check or test timeouts. --- .../beam/it/gcp/datastream/DatastreamResourceManager.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java index 8fc52ac4bf..0ab1e58a45 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java @@ -376,7 +376,9 @@ public synchronized DestinationConfig buildGCSDestinationConfig( DestinationConfig.newBuilder().setDestinationConnectionProfile(connectionProfile.getName()); GcsDestinationConfig.Builder gcsDestinationConfigBuilder = - GcsDestinationConfig.newBuilder().setPath(path); + GcsDestinationConfig.newBuilder() + .setPath(path) + .setFileRotationInterval(Duration.newBuilder().setSeconds(60).build()); if (destinationOutputFormat == DestinationOutputFormat.AVRO_FILE_FORMAT) { gcsDestinationConfigBuilder.setAvroFileFormat(AvroFileFormat.getDefaultInstance()); From 7250a5fc526bd78446d2c9b4fd46188e9c760d99 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Wed, 19 Aug 2026 10:52:45 -0400 Subject: [PATCH 2/4] [test only] add logs --- .../datastream/DatastreamResourceManager.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java index 0ab1e58a45..e9738b1592 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java @@ -639,11 +639,12 @@ public synchronized Stream pauseStream(Stream stream) { } public synchronized void cleanupAll() { - LOG.info("Cleaning up Datastream resource manager."); + LOG.info("Cleaning up Datastream resource manager at {}.", java.time.Instant.now()); boolean producedError = false; for (String stream : createdStreamIds) { try { + LOG.info("Starting deleteStreamAsync get for {} at {}", stream, java.time.Instant.now()); Failsafe.with(retryOnException()) .get( () -> @@ -653,6 +654,7 @@ public synchronized void cleanupAll() { .setName(StreamName.format(projectId, location, stream)) .build()) .get()); + LOG.info("Finished deleteStreamAsync get for {} at {}", stream, java.time.Instant.now()); } catch (Exception e) { if (ExceptionUtils.containsType(e, NotFoundException.class) || ExceptionUtils.containsMessage(e, "NOT_FOUND")) { @@ -664,10 +666,14 @@ public synchronized void cleanupAll() { } } } - LOG.info("Successfully deleted stream(s). "); + LOG.info("Successfully deleted stream(s) at {}. ", java.time.Instant.now()); for (String connectionProfile : createdConnectionProfileIds) { try { + LOG.info( + "Starting deleteConnectionProfileAsync get for {} at {}", + connectionProfile, + java.time.Instant.now()); Failsafe.with(retryOnException()) .get( () -> @@ -679,6 +685,10 @@ public synchronized void cleanupAll() { projectId, location, connectionProfile)) .build()) .get()); + LOG.info( + "Finished deleteConnectionProfileAsync get for {} at {}", + connectionProfile, + java.time.Instant.now()); } catch (Exception e) { if (ExceptionUtils.containsType(e, NotFoundException.class) || ExceptionUtils.containsMessage(e, "NOT_FOUND")) { @@ -692,7 +702,7 @@ public synchronized void cleanupAll() { } } } - LOG.info("Successfully deleted connection profile(s). "); + LOG.info("Successfully deleted connection profile(s) at {}. ", java.time.Instant.now()); try { datastreamClient.close(); From 174ebcf55c74a880ad7b1ceeea37bc6eccffdf3e Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 21 Aug 2026 16:28:14 -0400 Subject: [PATCH 3/4] Reduce streaming pipeline timeout & parallelize connection profile deletion --- .../datastream/DatastreamResourceManager.java | 83 ++++++++++--------- .../v2/templates/DataStreamToBigQueryIT.java | 2 +- 2 files changed, 46 insertions(+), 39 deletions(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java index e9738b1592..ade6779a8b 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datastream/DatastreamResourceManager.java @@ -69,6 +69,7 @@ import java.util.Set; import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.arrow.util.VisibleForTesting; import org.apache.beam.it.common.ResourceManager; import org.apache.beam.it.common.utils.ExceptionUtils; @@ -640,7 +641,7 @@ public synchronized Stream pauseStream(Stream stream) { public synchronized void cleanupAll() { LOG.info("Cleaning up Datastream resource manager at {}.", java.time.Instant.now()); - boolean producedError = false; + AtomicBoolean producedError = new AtomicBoolean(false); for (String stream : createdStreamIds) { try { @@ -662,56 +663,62 @@ public synchronized void cleanupAll() { "Stream {} not found in project {}. Assuming already deleted.", stream, projectId); } else { LOG.error("Failed to delete stream {}.", stream, e); - producedError = true; + producedError.set(true); } } } LOG.info("Successfully deleted stream(s) at {}. ", java.time.Instant.now()); - for (String connectionProfile : createdConnectionProfileIds) { - try { - LOG.info( - "Starting deleteConnectionProfileAsync get for {} at {}", - connectionProfile, - java.time.Instant.now()); - Failsafe.with(retryOnException()) - .get( - () -> - datastreamClient - .deleteConnectionProfileAsync( - DeleteConnectionProfileRequest.newBuilder() - .setName( - ConnectionProfileName.format( - projectId, location, connectionProfile)) - .build()) - .get()); - LOG.info( - "Finished deleteConnectionProfileAsync get for {} at {}", - connectionProfile, - java.time.Instant.now()); - } catch (Exception e) { - if (ExceptionUtils.containsType(e, NotFoundException.class) - || ExceptionUtils.containsMessage(e, "NOT_FOUND")) { - LOG.warn( - "Connection Profile {} not found in project {}. Assuming already deleted.", - connectionProfile, - projectId); - } else { - LOG.error("Failed to delete connection profile {}.", connectionProfile, e); - producedError = true; - } - } - } + createdConnectionProfileIds.parallelStream() + .forEach( + connectionProfile -> { + try { + LOG.info( + "Starting deleteConnectionProfileAsync get for {} at {}", + connectionProfile, + java.time.Instant.now()); + Failsafe.with(retryOnException()) + .get( + () -> + datastreamClient + .deleteConnectionProfileAsync( + DeleteConnectionProfileRequest.newBuilder() + .setName( + ConnectionProfileName.format( + projectId, location, connectionProfile)) + .build()) + .get()); + LOG.info( + "Finished deleteConnectionProfileAsync get for {} at {}", + connectionProfile, + java.time.Instant.now()); + } catch (Exception e) { + if (ExceptionUtils.containsType(e, NotFoundException.class) + || ExceptionUtils.containsMessage(e, "NOT_FOUND")) { + LOG.warn( + "Connection Profile {} not found in project {}. Assuming already deleted.", + connectionProfile, + projectId); + } else { + LOG.error( + "Failed to delete connection profile {} at {}.", + connectionProfile, + java.time.Instant.now(), + e); + producedError.set(true); + } + } + }); LOG.info("Successfully deleted connection profile(s) at {}. ", java.time.Instant.now()); try { datastreamClient.close(); } catch (Exception e) { LOG.error("Failed to close datastream client. "); - producedError = true; + producedError.set(true); } - if (producedError) { + if (producedError.get()) { throw new DatastreamResourceManagerException( "Failed to delete resources. Check above for errors."); } diff --git a/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java b/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java index 5bc48792d4..e355ba8ea4 100644 --- a/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java +++ b/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java @@ -393,7 +393,7 @@ private void simpleJdbcToBigQueryTest( // Job needs to be cancelled as draining will time out PipelineOperator.Result result = pipelineOperator() - .waitForConditionAndCancel(createConfig(info, Duration.ofMinutes(25)), conditionCheck); + .waitForConditionAndCancel(createConfig(info, Duration.ofMinutes(20)), conditionCheck); // Assert checkBigQueryTable(tableName, cdcEvents); From db7dd038861efdb1f53986ad78eed04042fdcfcc Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 21 Aug 2026 17:36:46 -0400 Subject: [PATCH 4/4] Move datastream resource manager to class level --- .../v2/templates/DataStreamToBigQueryIT.java | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java b/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java index e355ba8ea4..0874fd0411 100644 --- a/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java +++ b/v2/datastream-to-bigquery/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQueryIT.java @@ -67,6 +67,7 @@ import org.apache.commons.lang3.RandomStringUtils; import org.checkerframework.checker.nullness.qual.NonNull; import org.junit.After; +import org.junit.AfterClass; import org.junit.Before; import org.junit.Ignore; import org.junit.Rule; @@ -103,16 +104,19 @@ enum JDBCType { private CloudSqlResourceManager cloudSqlResourceManager; private PubsubResourceManager pubsubResourceManager; - private DatastreamResourceManager datastreamResourceManager; + private static DatastreamResourceManager datastreamResourceManager; private BigQueryResourceManager bigQueryResourceManager; @Before public void setUp() throws IOException { - datastreamResourceManager = - DatastreamResourceManager.builder(testName, PROJECT, REGION) - .setCredentialsProvider(credentialsProvider) - .setPrivateConnectivity("datastream-connect-2") - .build(); + if (datastreamResourceManager == null) { + datastreamResourceManager = + DatastreamResourceManager.builder( + DataStreamToBigQueryIT.class.getSimpleName(), PROJECT, REGION) + .setCredentialsProvider(credentialsProvider) + .setPrivateConnectivity("datastream-connect-2") + .build(); + } bigQueryResourceManager = BigQueryResourceManager.builder(testName, PROJECT, credentials).build(); @@ -143,10 +147,12 @@ public void setUp() throws IOException { @After public void cleanUp() { ResourceManagerUtils.cleanResources( - cloudSqlResourceManager, - pubsubResourceManager, - datastreamResourceManager, - bigQueryResourceManager); + cloudSqlResourceManager, pubsubResourceManager, bigQueryResourceManager); + } + + @AfterClass + public static void tearDownClass() { + ResourceManagerUtils.cleanResources(datastreamResourceManager); } @Test @@ -393,7 +399,7 @@ private void simpleJdbcToBigQueryTest( // Job needs to be cancelled as draining will time out PipelineOperator.Result result = pipelineOperator() - .waitForConditionAndCancel(createConfig(info, Duration.ofMinutes(20)), conditionCheck); + .waitForConditionAndCancel(createConfig(info, Duration.ofMinutes(25)), conditionCheck); // Assert checkBigQueryTable(tableName, cdcEvents);