Fix TestPipeline.runWithAdditionalOptionArgs ignore additionalArgs (#25937)

* Fix TestPipeline.runWithAdditionalOptionArgs ignore additionalArgs
  when system property PROPERTY_BEAM_TEST_PIPELINE_OPTIONS is empty

Author: Abacn

sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java

@@ -347,29 +347,28 @@ public PipelineResult runWithAdditionalOptionArgs(List<String> additionalArgs) {
     try {
       @Nullable
       String beamTestPipelineOptions = System.getProperty(PROPERTY_BEAM_TEST_PIPELINE_OPTIONS);
-      PipelineOptions options;
-      if (Strings.isNullOrEmpty(beamTestPipelineOptions)) {
-        options = PipelineOptionsFactory.create();
-      } else {
-        List<String> args = MAPPER.readValue(beamTestPipelineOptions, List.class);
-        args.addAll(additionalArgs);
-        String[] newArgs = new String[args.size()];
-        newArgs = args.toArray(newArgs);
-        options = PipelineOptionsFactory.fromArgs(newArgs).as(TestPipelineOptions.class);
+      List<String> args = new ArrayList<>();
+      if (!Strings.isNullOrEmpty(beamTestPipelineOptions)) {
+        args.addAll(MAPPER.readValue(beamTestPipelineOptions, List.class));
       }
+      args.addAll(additionalArgs);
+      String[] newArgs = new String[args.size()];
+      newArgs = args.toArray(newArgs);
+      PipelineOptions newOptions =
+          PipelineOptionsFactory.fromArgs(newArgs).as(TestPipelineOptions.class);
 
       // If no options were specified, set some reasonable defaults
       if (Strings.isNullOrEmpty(beamTestPipelineOptions)) {
         // If there are no provided options, check to see if a dummy runner should be used.
         String useDefaultDummy = System.getProperty(PROPERTY_USE_DEFAULT_DUMMY_RUNNER);
         if (!Strings.isNullOrEmpty(useDefaultDummy) && Boolean.valueOf(useDefaultDummy)) {
-          options.setRunner(CrashingRunner.class);
+          newOptions.setRunner(CrashingRunner.class);
         }
       }
-      options.setStableUniqueNames(CheckEnabled.ERROR);
+      newOptions.setStableUniqueNames(CheckEnabled.ERROR);
 
       FileSystems.setDefaultPipelineOptions(options);
-      return run(options);
+      return run(newOptions);
     } catch (IOException e) {
       throw new RuntimeException(
           "Unable to instantiate test options from system property "

sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java

@@ -25,6 +25,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
@@ -37,9 +38,11 @@
 import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.SimpleFunction;
 import org.apache.beam.sdk.util.common.ReflectHelpers;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.hamcrest.BaseMatcher;
 import org.hamcrest.Description;
@@ -270,6 +273,29 @@ public void testDanglingPAssertValidatesRunner() throws Exception {
         PAssert.that(pCollection).containsInAnyOrder(WHATEVER);
       }
 
+      @Category(NeedsRunner.class)
+      @Test
+      public void testRunWithAdditionalArgsEffective() {
+        ArrayList<String> pipelineArgs = new ArrayList<String>();
+        pipelineArgs.add("--tempLocation=gs://some-location/");
+        pipeline.apply(Create.of("")).apply(new ValidateTempLocation<>());
+        PipelineResult.State result =
+            pipeline.runWithAdditionalOptionArgs(pipelineArgs).waitUntilFinish();
+        assert (result == PipelineResult.State.DONE);
+      }
+
+      static class ValidateTempLocation<T> extends PTransform<PCollection<T>, PCollection<T>> {
+        @Override
+        public void validate(PipelineOptions pipelineOptions) {
+          assert (!Strings.isNullOrEmpty(pipelineOptions.getTempLocation()));
+        }
+
+        @Override
+        public PCollection<T> expand(PCollection<T> input) {
+          return input;
+        }
+      }
+
       /**
        * Tests that a {@link TestPipeline} rule behaves as expected when there is no pipeline usage
        * within a test that has a {@link ValidatesRunner} annotation.