From b7809c2fa704e8418389714f49de7632639c83d1 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 13 Nov 2026 00:12:10 +0000 Subject: [PATCH 1/2] Re-stage Dataflow packages older than 1 day in PackageUtil --- .../runners/dataflow/util/PackageUtil.java | 29 +++++++++++++++++-- .../dataflow/util/PackageUtilTest.java | 28 ++++++++++++++++++ 2 files changed, 54 insertions(+), 3 deletions(-) diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java index 360fd06d7bd0..f938bb0f806b 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java @@ -47,6 +47,7 @@ import org.apache.beam.sdk.extensions.gcp.util.BackOffAdapter; import org.apache.beam.sdk.io.FileSystems; import org.apache.beam.sdk.io.fs.CreateOptions; +import org.apache.beam.sdk.io.fs.MatchResult.Metadata; import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions; import org.apache.beam.sdk.util.FluentBackoff; import org.apache.beam.sdk.util.MimeTypes; @@ -122,11 +123,33 @@ private CompletionStage computePackageAttributes( () -> PackageAttributes.forFileToStage(source, hash, dest, stagingPath), executorService); } + /** + * Maximum age of a previously staged file before it is re-uploaded. Shared test and staging + * buckets often configure GCS lifecycle deletion rules (for example, 7 days on {@code + * gs://temp-storage-for-validates-runner-tests}), so re-uploading files older than 1 day + * refreshes their object creation timestamp well before lifecycle deletion can race with job + * execution. + */ + static final Duration MAX_STAGED_FILE_AGE = Duration.standardDays(1); + private boolean alreadyStaged(PackageAttributes attributes) throws IOException { try { - long remoteLength = - FileSystems.matchSingleFileSpec(attributes.getDestination().getLocation()).sizeBytes(); - return remoteLength == attributes.getSize(); + Metadata remoteMetadata = + FileSystems.matchSingleFileSpec(attributes.getDestination().getLocation()); + if (remoteMetadata.sizeBytes() != attributes.getSize()) { + return false; + } + long lastModifiedMillis = remoteMetadata.lastModifiedMillis(); + if (lastModifiedMillis > 0 + && Instant.now().getMillis() - lastModifiedMillis > MAX_STAGED_FILE_AGE.getMillis()) { + LOG.info( + "Re-staging file {} at {} because existing object is older than {}", + attributes.getSourceDescription(), + attributes.getDestination().getLocation(), + MAX_STAGED_FILE_AGE); + return false; + } + return true; } catch (FileNotFoundException expected) { // If the file doesn't exist, it means we need to upload it. return false; diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java index 7e7ef6a8345a..8f67fc54944b 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java @@ -49,6 +49,7 @@ import com.google.api.client.testing.http.MockHttpTransport; import com.google.api.client.testing.http.MockLowLevelHttpRequest; import com.google.api.client.testing.http.MockLowLevelHttpResponse; +import com.google.api.client.util.DateTime; import com.google.api.services.dataflow.model.DataflowPackage; import com.google.api.services.storage.model.StorageObject; import java.io.File; @@ -91,6 +92,8 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors; import org.checkerframework.checker.nullness.qual.Nullable; import org.hamcrest.Matchers; +import org.joda.time.Duration; +import org.joda.time.Instant; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -541,6 +544,31 @@ public void testPackageUploadIsNotSkippedWhenSizesAreDifferent() throws Exceptio verifyNoMoreInteractions(mockGcsUtil); } + @Test + public void testPackageUploadIsNotSkippedWhenStagedFileIsExpired() throws Exception { + Pipe pipe = Pipe.open(); + File tmpFile = makeFileWithContents("file.txt", "This is a test!"); + StorageObject expiredObject = + createStorageObject(STAGING_PATH, tmpFile.length()) + .setUpdated( + new DateTime( + Instant.now() + .minus(PackageUtil.MAX_STAGED_FILE_AGE) + .minus(Duration.standardMinutes(1)) + .getMillis())); + when(mockGcsUtil.getObjects(anyList())) + .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(expiredObject))); + when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class))) + .thenReturn(pipe.sink()); + + defaultPackageUtil.stageClasspathElements( + ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())), STAGING_PATH, createOptions); + + verify(mockGcsUtil).getObjects(anyList()); + verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)); + verifyNoMoreInteractions(mockGcsUtil); + } + @Test public void testPackageUploadWithExplicitPackageName() throws Exception { Pipe pipe = Pipe.open(); From 524a9ee2757e8503a0cbc44b67b3c3ad62d65f8c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 13 Nov 2026 05:26:29 +0000 Subject: [PATCH 2/2] Scope shared TestDataflowRunner staging directory by UTC date instead of modifying PackageUtil --- .../runners/dataflow/TestDataflowRunner.java | 21 ++++++++++++-- .../runners/dataflow/util/PackageUtil.java | 29 ++----------------- .../dataflow/TestDataflowRunnerTest.java | 14 ++++++++- .../dataflow/util/PackageUtilTest.java | 28 ------------------ 4 files changed, 35 insertions(+), 57 deletions(-) diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java index b404ceaf41dc..23819a0a69c4 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java @@ -32,6 +32,8 @@ import java.io.File; import java.io.IOException; import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.ZoneOffset; import java.util.List; import java.util.Map; import java.util.concurrent.Callable; @@ -70,6 +72,10 @@ }) public class TestDataflowRunner extends PipelineRunner { private static final String TENTATIVE_COUNTER = "tentative"; + + @VisibleForTesting + static final String DAILY_STAGING_DIR = "staging-" + LocalDate.now(ZoneOffset.UTC); + static final String MAX_CONCURRENT_STANDALONE_JOBS_PROPERTY = "beam.dataflow.maxConcurrentStandaloneJobs"; private static final int DEFAULT_MAX_CONCURRENT_STANDALONE_JOBS = 12; @@ -148,9 +154,20 @@ public static TestDataflowRunner fromOptions(PipelineOptions options) { dataflowOptions.setTempLocation(tempLocation); String defaultPerJobStagingLocation = FileSystems.matchNewDirectory(tempLocation, "staging").toString(); - if (defaultPerJobStagingLocation.equals(dataflowOptions.getStagingLocation())) { + String unversionedSharedStagingLocation = + FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), "staging").toString(); + String rawStagingLocation = dataflowOptions.getStagingLocation(); + String configuredStagingLocation = + rawStagingLocation != null + ? FileSystems.matchNewDirectory(rawStagingLocation).toString() + : null; + // Scope shared test staging directories by UTC date so tests running on the same day reuse + // staged JARs without referencing objects near the 7-day GCS lifecycle deletion threshold. + if (defaultPerJobStagingLocation.equals(configuredStagingLocation) + || unversionedSharedStagingLocation.equals(configuredStagingLocation)) { dataflowOptions.setStagingLocation( - FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), "staging").toString()); + FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), DAILY_STAGING_DIR) + .toString()); } return new TestDataflowRunner( diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java index f938bb0f806b..360fd06d7bd0 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java @@ -47,7 +47,6 @@ import org.apache.beam.sdk.extensions.gcp.util.BackOffAdapter; import org.apache.beam.sdk.io.FileSystems; import org.apache.beam.sdk.io.fs.CreateOptions; -import org.apache.beam.sdk.io.fs.MatchResult.Metadata; import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions; import org.apache.beam.sdk.util.FluentBackoff; import org.apache.beam.sdk.util.MimeTypes; @@ -123,33 +122,11 @@ private CompletionStage computePackageAttributes( () -> PackageAttributes.forFileToStage(source, hash, dest, stagingPath), executorService); } - /** - * Maximum age of a previously staged file before it is re-uploaded. Shared test and staging - * buckets often configure GCS lifecycle deletion rules (for example, 7 days on {@code - * gs://temp-storage-for-validates-runner-tests}), so re-uploading files older than 1 day - * refreshes their object creation timestamp well before lifecycle deletion can race with job - * execution. - */ - static final Duration MAX_STAGED_FILE_AGE = Duration.standardDays(1); - private boolean alreadyStaged(PackageAttributes attributes) throws IOException { try { - Metadata remoteMetadata = - FileSystems.matchSingleFileSpec(attributes.getDestination().getLocation()); - if (remoteMetadata.sizeBytes() != attributes.getSize()) { - return false; - } - long lastModifiedMillis = remoteMetadata.lastModifiedMillis(); - if (lastModifiedMillis > 0 - && Instant.now().getMillis() - lastModifiedMillis > MAX_STAGED_FILE_AGE.getMillis()) { - LOG.info( - "Re-staging file {} at {} because existing object is older than {}", - attributes.getSourceDescription(), - attributes.getDestination().getLocation(), - MAX_STAGED_FILE_AGE); - return false; - } - return true; + long remoteLength = + FileSystems.matchSingleFileSpec(attributes.getDestination().getLocation()).sizeBytes(); + return remoteLength == attributes.getSize(); } catch (FileNotFoundException expected) { // If the file doesn't exist, it means we need to upload it. return false; diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java index 6e1d9f627ffa..0fb4b0eb6f3e 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java @@ -119,7 +119,19 @@ public void testFromOptionsUsesSharedStagingLocationUnderTempRoot() { TestDataflowRunner.fromOptions(options); assertEquals("gs://test/test-job-1/output/results", options.getTempLocation()); assertEquals("gs://test/test-job-1/output/results", options.getGcpTempLocation()); - assertEquals("gs://test/staging/", options.getStagingLocation()); + assertEquals( + "gs://test/" + TestDataflowRunner.DAILY_STAGING_DIR + "/", options.getStagingLocation()); + } + + @Test + public void testFromOptionsScopesUnversionedTempRootStagingLocationByDate() { + options.setJobName("test-job-unversioned-staging"); + options.setStagingLocation("gs://test/staging"); + TestDataflowRunner.fromOptions(options); + assertEquals( + "gs://test/test-job-unversioned-staging/output/results", options.getTempLocation()); + assertEquals( + "gs://test/" + TestDataflowRunner.DAILY_STAGING_DIR + "/", options.getStagingLocation()); } @Test diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java index 8f67fc54944b..7e7ef6a8345a 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java @@ -49,7 +49,6 @@ import com.google.api.client.testing.http.MockHttpTransport; import com.google.api.client.testing.http.MockLowLevelHttpRequest; import com.google.api.client.testing.http.MockLowLevelHttpResponse; -import com.google.api.client.util.DateTime; import com.google.api.services.dataflow.model.DataflowPackage; import com.google.api.services.storage.model.StorageObject; import java.io.File; @@ -92,8 +91,6 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors; import org.checkerframework.checker.nullness.qual.Nullable; import org.hamcrest.Matchers; -import org.joda.time.Duration; -import org.joda.time.Instant; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -544,31 +541,6 @@ public void testPackageUploadIsNotSkippedWhenSizesAreDifferent() throws Exceptio verifyNoMoreInteractions(mockGcsUtil); } - @Test - public void testPackageUploadIsNotSkippedWhenStagedFileIsExpired() throws Exception { - Pipe pipe = Pipe.open(); - File tmpFile = makeFileWithContents("file.txt", "This is a test!"); - StorageObject expiredObject = - createStorageObject(STAGING_PATH, tmpFile.length()) - .setUpdated( - new DateTime( - Instant.now() - .minus(PackageUtil.MAX_STAGED_FILE_AGE) - .minus(Duration.standardMinutes(1)) - .getMillis())); - when(mockGcsUtil.getObjects(anyList())) - .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(expiredObject))); - when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class))) - .thenReturn(pipe.sink()); - - defaultPackageUtil.stageClasspathElements( - ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())), STAGING_PATH, createOptions); - - verify(mockGcsUtil).getObjects(anyList()); - verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)); - verifyNoMoreInteractions(mockGcsUtil); - } - @Test public void testPackageUploadWithExplicitPackageName() throws Exception { Pipe pipe = Pipe.open();