Copy experiments list to mutable ArrayList before addExperiment() calls

Author: ash6898

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java

@@ -403,6 +403,10 @@ public Translator(Pipeline pipeline, DataflowRunner runner, SdkComponents sdkCom
      * @return a Job definition filled in with the type of job, the environment, and the job steps.
      */
     public Job translate(List<DataflowPackage> packages) {
+      // Ensure the experiments list is mutable before any experiments are added.
+      if (options.getExperiments() != null) {
+        options.setExperiments(new ArrayList<>(options.getExperiments()));
+      }
       job.setName(options.getJobName().toLowerCase());
 
       Environment environment = new Environment();

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

@@ -1241,15 +1241,21 @@ private static boolean includesTransformUpgrades(Pipeline pipeline) {
   @SuppressWarnings("Slf4jFormatShouldBeConst")
   @Override
   public DataflowPipelineJob run(Pipeline pipeline) {
+    // Ensure the experiments list is mutable before any experiments are added.
+    if (options.getExperiments() != null) {
+      options.setExperiments(new ArrayList<>(options.getExperiments()));
+    }
     // Multi-language pipelines and pipelines that include upgrades should automatically be upgraded
     // to Dataflow Portable Runner.
     if (DataflowRunner.isMultiLanguagePipeline(pipeline) || includesTransformUpgrades(pipeline)) {
-      if (!firstNonNull(options.getExperiments(), Collections.emptyList())
-          .contains("use_runner_v2")) {
-        LOG.info(
-            "Automatically enabling Dataflow Portable Runner since the pipeline used cross-language"
-                + " transforms or pipeline needed a transform upgrade.");
-        ExperimentalOptions.addExperiment(options, "use_runner_v2");
+      if (!useUnifiedWorker(options)) {
+        if (!firstNonNull(options.getExperiments(), Collections.emptyList())
+            .contains("use_runner_v2")) {
+          LOG.info(
+              "Automatically enabling Dataflow Portable Runner since the pipeline used cross-language"
+                  + " transforms or pipeline needed a transform upgrade.");
+          ExperimentalOptions.addExperiment(options, "use_runner_v2");
+        }
       }
     }
     if (useUnifiedWorker(options)) {

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcWindmillServer.java

@@ -111,6 +111,10 @@ private static DataflowWorkerHarnessOptions testOptions(
       boolean enableStreamingEngine, List<String> additionalExperiments) {
     DataflowWorkerHarnessOptions options =
         PipelineOptionsFactory.create().as(DataflowWorkerHarnessOptions.class);
+    // Ensure the experiments list is mutable before any experiments are added.
+    if (options.getExperiments() != null) {
+      options.setExperiments(new ArrayList<>(options.getExperiments()));
+    }
     options.setProject("project");
     options.setJobId("job");
     options.setWorkerId("worker");