Attempt to fix PostCommit PVR Spark

apache · Oct 9, 2024 · 482a93c · 482a93c
1 parent 42cad40
commit 482a93c
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 4 deletions.
diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark_Batch.json
@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 0
+  "modification": 1
 }
diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -2587,6 +2587,8 @@ class BeamModulePlugin implements Plugin<Project> {
           def javaContainerSuffix = ver ? "java$ver" : getSupportedJavaVersion()
           dependsOn ":sdks:java:container:${javaContainerSuffix}:docker"
         }
+        outputs.upToDateWhen {false}
+        jvmArgs = ["-javaagent:/Users/yathu/beam/file-leak-detector-1.13-jar-with-dependencies.jar=threshold=100"]
       }
     }
 

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileSystems.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileSystems.java
@@ -565,7 +565,12 @@ static FileSystem getFileSystemInternal(String scheme) {
    *
    * <p>It will be used in {@link FileSystemRegistrar FileSystemRegistrars} for all schemes.
    *
-   * <p>This is expected only to be used by runners after {@code Pipeline.run}, or in tests.
+   * <p>This is expected only to be used by runners after {@code Pipeline.run}. User code executed
+   * during pipeline execution time should not call it.
+   *
+   * <p>Outside pipeline execution environments where Beam FileSystem API is used (e.g. test
+   * methods, user code executed during pipeline submission), use
+   * {@link #setDefaultPipelineOptionsOnce} instead.
    */
   @Internal
   public static void setDefaultPipelineOptions(PipelineOptions options) {
@@ -593,6 +598,20 @@ public static void setDefaultPipelineOptions(PipelineOptions options) {
     }
   }
 
+  /**
+   * Sets the default configuration once. Used by test host (e.g. JUnit test class).
+   *
+   * <p>This method executes setDefaultPipelineOptions if SCHEME_TO_FILESYSTEM is empty. This avoids
+   * conflicts in repeated calls to setDefaultPipelineOptions for every TestPipeline JVM runtime
+   * constructing.
+   */
+  @Internal
+  public static synchronized void setDefaultPipelineOptionsOnce(PipelineOptions options) {
+    if (SCHEME_TO_FILESYSTEM.get().isEmpty()) {
+      setDefaultPipelineOptions(options);
+    }
+  }
+
   @VisibleForTesting
   static Map<String, FileSystem> verifySchemesAreUnique(
       PipelineOptions options, Set<FileSystemRegistrar> registrars) {

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
@@ -373,7 +373,7 @@ public PipelineResult runWithAdditionalOptionArgs(List<String> additionalArgs) {
       }
       newOptions.setStableUniqueNames(CheckEnabled.ERROR);
 
-      FileSystems.setDefaultPipelineOptions(options);
+      FileSystems.setDefaultPipelineOptionsOnce(options);
       return run(newOptions);
     } catch (IOException e) {
       throw new RuntimeException(
@@ -515,7 +515,7 @@ public static PipelineOptions testingPipelineOptions() {
       }
       options.setStableUniqueNames(CheckEnabled.ERROR);
 
-      FileSystems.setDefaultPipelineOptions(options);
+      FileSystems.setDefaultPipelineOptionsOnce(options);
       return options;
     } catch (IOException e) {
       throw new RuntimeException(