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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
