You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "reuvenlax (via GitHub)" <gi...@apache.org> on 2026/09/28 19:04:09 UTC
[PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
reuvenlax opened a new pull request, #40321:
URL: https://github.com/apache/beam/pull/40321
TestDataflowRunner previously set tempLocation to <tempRoot>/<jobName>/output/results, causing DataflowPipelineOptions.StagingLocationFactory` to derive a unique path for every single test job. Consequently, every test re-hashed ~300+ MB of classpath jars and re-queried/uploaded them to a brand-new GCS staging directory instead of reusing staged artifacts across tests.
**Fix**: Defaults stagingLocation to a shared /staging/ directory across jobs in the same test run (while keeping tempLocation isolated per job at <tempRoot>/<jobName>/output/results).
Caches SHA-256 file hashes and directory zip artifacts in memory keyed by (path, size, lastModifiedMillis) so unmodified classpath jars are hashed once per worker JVM instead of once per test.
Caches resolved DataflowPackage lists per `stagingLocation, filesToStage) when running under TestDataflowRunner so each worker JVM stages/verifies the classpath once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4135374809
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -49,6 +67,16 @@ public static GcsStager fromOptions(PipelineOptions options) {
*/
@Override
public List<DataflowPackage> stageFiles(List<StagedFile> filesToStage) {
+ String stagingLocation = options.getStagingLocation();
+ if (stagingLocation != null && TestDataflowRunner.class.equals(options.getRunner())) {
Review Comment:
unlikely to help for other runners (since it's rare to reuse the same JVM for multiple pipelines), however sure - removed this guard.
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -30,6 +34,20 @@
/** Utility class for staging files to GCS. */
public class GcsStager implements Stager {
+ @AutoValue
+ abstract static class StagedFilesCacheKey {
+ abstract String getStagingLocation();
+
+ abstract List<StagedFile> getFilesToStage();
+
+ static StagedFilesCacheKey of(String stagingLocation, List<StagedFile> filesToStage) {
+ return new AutoValue_GcsStager_StagedFilesCacheKey(stagingLocation, filesToStage);
+ }
+ }
+
+ private static final ConcurrentHashMap<StagedFilesCacheKey, List<DataflowPackage>>
+ STAGED_FILES_CACHE = new ConcurrentHashMap<>();
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4126508552
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java:
##########
@@ -102,6 +102,24 @@ public void testToString() {
"TestDataflowRunner#TestAppName", TestDataflowRunner.fromOptions(options).toString());
}
+ @Test
Review Comment:
taking a look....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4135452885
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -49,6 +67,16 @@ public static GcsStager fromOptions(PipelineOptions options) {
*/
@Override
public List<DataflowPackage> stageFiles(List<StagedFile> filesToStage) {
+ String stagingLocation = options.getStagingLocation();
+ if (stagingLocation != null && TestDataflowRunner.class.equals(options.getRunner())) {
Review Comment:
fewer deps and lower cyclomatic complexity are also worthwhile :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax merged PR #40321:
URL: https://github.com/apache/beam/pull/40321
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4126587444
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -49,6 +56,16 @@ public static GcsStager fromOptions(PipelineOptions options) {
*/
@Override
public List<DataflowPackage> stageFiles(List<StagedFile> filesToStage) {
+ String stagingLocation = options.getStagingLocation();
+ if (stagingLocation != null && TestDataflowRunner.class.equals(options.getRunner())) {
+ List<Object> cacheKey = Arrays.asList(stagingLocation, filesToStage);
Review Comment:
Done
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java:
##########
@@ -102,6 +102,24 @@ public void testToString() {
"TestDataflowRunner#TestAppName", TestDataflowRunner.fromOptions(options).toString());
}
+ @Test
Review Comment:
it's because when doing this only TestDataflowRunner.fromOptions(options) is called and never Pipeline.create(options). Pipeline.create(options) registers the filesystems as a side effect.
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40321:
URL: https://github.com/apache/beam/pull/40321#issuecomment-5893238575
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40321:
URL: https://github.com/apache/beam/pull/40321#issuecomment-5892567220
Assigning reviewers:
R: @tvalentyn added as fallback since no labels match configuration
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on PR #40321:
URL: https://github.com/apache/beam/pull/40321#issuecomment-5893225971
R: @kennknowles
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4126489904
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java:
##########
@@ -102,6 +102,24 @@ public void testToString() {
"TestDataflowRunner#TestAppName", TestDataflowRunner.fromOptions(options).toString());
}
+ @Test
Review Comment:
(Not directly due to this PR) Running the test alone (`./gradlew :runners:google-cloud-dataflow-java:test --tests *testFromOptionsUsesSharedStagingLocationUnderTempRoot`) it fails with
```
java.lang.IllegalArgumentException: No filesystem found for scheme gs
at org.apache.beam.sdk.io.FileSystems.getFileSystemInternal(FileSystems.java:558)
at org.apache.beam.sdk.io.FileSystems.matchNewResource(FileSystems.java:665)
at org.apache.beam.sdk.io.FileSystems.matchNewDirectory(FileSystems.java:678)
at org.apache.beam.runners.dataflow.TestDataflowRunner.fromOptions(TestDataflowRunner.java:78)
at org.apache.beam.runners.dataflow.TestDataflowRunnerTest.testFromOptionsUsesSharedStagingLocationUnderTempRoot(TestDataflowRunnerTest.java:108)
```
but running the test class passed. Likely a pre-existing issue though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4126343671
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -49,6 +56,16 @@ public static GcsStager fromOptions(PipelineOptions options) {
*/
@Override
public List<DataflowPackage> stageFiles(List<StagedFile> filesToStage) {
+ String stagingLocation = options.getStagingLocation();
+ if (stagingLocation != null && TestDataflowRunner.class.equals(options.getRunner())) {
+ List<Object> cacheKey = Arrays.asList(stagingLocation, filesToStage);
Review Comment:
since the key is a pair of (String, List<StagedFile>), consider use it as STAGED_FILES_CACHE's key type. One can use guava's com.google.common.collect.Maps.immutableEntry or org.apache.commons.lang3.tuple.Pair
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make ValidatesRunner faster: Share and cache GCS staging across TestDataflowRunner executions [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #40321:
URL: https://github.com/apache/beam/pull/40321#discussion_r4134749978
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -30,6 +34,20 @@
/** Utility class for staging files to GCS. */
public class GcsStager implements Stager {
+ @AutoValue
+ abstract static class StagedFilesCacheKey {
+ abstract String getStagingLocation();
+
+ abstract List<StagedFile> getFilesToStage();
+
+ static StagedFilesCacheKey of(String stagingLocation, List<StagedFile> filesToStage) {
+ return new AutoValue_GcsStager_StagedFilesCacheKey(stagingLocation, filesToStage);
+ }
+ }
+
+ private static final ConcurrentHashMap<StagedFilesCacheKey, List<DataflowPackage>>
+ STAGED_FILES_CACHE = new ConcurrentHashMap<>();
Review Comment:
Maybe use a `Cache` with a size bound that will be big enough for any reasonable situation, but will prevent future weirdness if things change.
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/GcsStager.java:
##########
@@ -49,6 +67,16 @@ public static GcsStager fromOptions(PipelineOptions options) {
*/
@Override
public List<DataflowPackage> stageFiles(List<StagedFile> filesToStage) {
+ String stagingLocation = options.getStagingLocation();
+ if (stagingLocation != null && TestDataflowRunner.class.equals(options.getRunner())) {
Review Comment:
Why depend on the runner here? Is there a reason it cannot just be always on?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org