Use ExperimentalOptions.addExperiment() instead of get/set pattern in Java

Replaces getExperiments()/modify/setExperiments() boilerplate with
ExperimentalOptions.addExperiment() which handles null-init and deduplication.

Resolves #19347

Author: ash6898

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

@@ -69,6 +69,7 @@
 import org.apache.beam.sdk.coders.IterableCoder;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.options.ExperimentalOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.StreamingOptions;
 import org.apache.beam.sdk.runners.AppliedPTransform;
@@ -414,19 +415,8 @@ public Job translate(List<DataflowPackage> packages) {
       // back end as well.  If streaming engine is not enabled make sure the experiments are also
       // not enabled.
       if (options.isEnableStreamingEngine()) {
-        List<String> experiments = options.getExperiments();
-        if (experiments == null) {
-          experiments = new ArrayList<String>();
-        } else {
-          experiments = new ArrayList<String>(experiments);
-        }
-        if (!experiments.contains(GcpOptions.STREAMING_ENGINE_EXPERIMENT)) {
-          experiments.add(GcpOptions.STREAMING_ENGINE_EXPERIMENT);
-        }
-        if (!experiments.contains(GcpOptions.WINDMILL_SERVICE_EXPERIMENT)) {
-          experiments.add(GcpOptions.WINDMILL_SERVICE_EXPERIMENT);
-        }
-        options.setExperiments(experiments);
+        ExperimentalOptions.addExperiment(options, GcpOptions.STREAMING_ENGINE_EXPERIMENT);
+        ExperimentalOptions.addExperiment(options, GcpOptions.WINDMILL_SERVICE_EXPERIMENT);
       } else {
         List<String> experiments = options.getExperiments();
         if (experiments != null) {

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

@@ -1244,13 +1244,11 @@ public DataflowPipelineJob run(Pipeline pipeline) {
     // Multi-language pipelines and pipelines that include upgrades should automatically be upgraded
     // to Runner v2.
     if (DataflowRunner.isMultiLanguagePipeline(pipeline) || includesTransformUpgrades(pipeline)) {
-      List<String> experiments = firstNonNull(options.getExperiments(), Collections.emptyList());
-      if (!experiments.contains("use_runner_v2")) {
+      if (!firstNonNull(options.getExperiments(), Collections.emptyList()).contains("use_runner_v2")) {
         LOG.info(
             "Automatically enabling Dataflow Runner v2 since the pipeline used cross-language"
                 + " transforms or pipeline needed a transform upgrade.");
-        options.setExperiments(
-            ImmutableList.<String>builder().addAll(experiments).add("use_runner_v2").build());
+        ExperimentalOptions.addExperiment(options, "use_runner_v2");
       }
     }
     if (useUnifiedWorker(options)) {
@@ -1260,21 +1258,10 @@ public DataflowPipelineJob run(Pipeline pipeline) {
         throw new IllegalArgumentException(
             "Runner V2 both disabled and enabled: at least one of ['beam_fn_api', 'use_unified_worker', 'use_runner_v2', 'use_portable_job_submission'] is set and also one of ['disable_runner_v2', 'disable_runner_v2_until_2023', 'disable_prime_runner_v2'] is set.");
       }
-      List<String> experiments =
-          new ArrayList<>(options.getExperiments()); // non-null if useUnifiedWorker is true
-      if (!experiments.contains("use_runner_v2")) {
-        experiments.add("use_runner_v2");
-      }
-      if (!experiments.contains("use_unified_worker")) {
-        experiments.add("use_unified_worker");
-      }
-      if (!experiments.contains("beam_fn_api")) {
-        experiments.add("beam_fn_api");
-      }
-      if (!experiments.contains("use_portable_job_submission")) {
-        experiments.add("use_portable_job_submission");
-      }
-      options.setExperiments(ImmutableList.copyOf(experiments));
+      ExperimentalOptions.addExperiment(options, "use_runner_v2");
+      ExperimentalOptions.addExperiment(options, "use_unified_worker");
+      ExperimentalOptions.addExperiment(options, "beam_fn_api");
+      ExperimentalOptions.addExperiment(options, "use_portable_job_submission");
       // Ensure that logging via the FnApi is enabled
       options.as(SdkHarnessOptions.class).setEnableLogViaFnApi(true);
     }
@@ -1301,14 +1288,9 @@ public DataflowPipelineJob run(Pipeline pipeline) {
       options.setStreaming(true);
 
       {
-        List<String> experiments =
-            options.getExperiments() == null
-                ? new ArrayList<>()
-                : new ArrayList<>(options.getExperiments());
         // Experiment marking that the harness supports tag encoding v2
         // Backend will enable tag encoding v2 only if the harness supports it.
-        experiments.add("streaming_engine_state_tag_encoding_v2_supported");
-        options.setExperiments(ImmutableList.copyOf(experiments));
+        ExperimentalOptions.addExperiment(options, "streaming_engine_state_tag_encoding_v2_supported");
       }
 
       if (useUnifiedWorker(options)) {
@@ -1413,15 +1395,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
             pipeline, dataflowV1PipelineProto, dataflowV1Components, this, packages);
 
     if (!isNullOrEmpty(dataflowOptions.getDataflowWorkerJar()) && !useUnifiedWorker(options)) {
-      List<String> experiments =
-          firstNonNull(dataflowOptions.getExperiments(), Collections.emptyList());
-      if (!experiments.contains("use_staged_dataflow_worker_jar")) {
-        dataflowOptions.setExperiments(
-            ImmutableList.<String>builder()
-                .addAll(experiments)
-                .add("use_staged_dataflow_worker_jar")
-                .build());
-      }
+      ExperimentalOptions.addExperiment(options, "use_staged_dataflow_worker_jar");
     }
 
     Job newJob = jobSpecification.getJob();
@@ -1480,11 +1454,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
               .collect(Collectors.toList());
 
       if (minCpuFlags.isEmpty()) {
-        dataflowOptions.setExperiments(
-            ImmutableList.<String>builder()
-                .addAll(experiments)
-                .add("min_cpu_platform=" + dataflowOptions.getMinCpuPlatform())
-                .build());
+        ExperimentalOptions.addExperiment(dataflowOptions, "min_cpu_platform=" + dataflowOptions.getMinCpuPlatform());
       } else {
         LOG.warn(
             "Flag min_cpu_platform is defined in both top level PipelineOption, "
@@ -1521,19 +1491,16 @@ public DataflowPipelineJob run(Pipeline pipeline) {
     byte[] jobGraphBytes = DataflowPipelineTranslator.jobToString(newJob).getBytes(UTF_8);
     int jobGraphByteSize = jobGraphBytes.length;
     if (jobGraphByteSize >= CREATE_JOB_REQUEST_LIMIT_BYTES
-        && !hasExperiment(options, "upload_graph")
         && !useUnifiedWorker(options)) {
-      List<String> experiments = firstNonNull(options.getExperiments(), Collections.emptyList());
-      options.setExperiments(
-          ImmutableList.<String>builder().addAll(experiments).add("upload_graph").build());
+      ExperimentalOptions.addExperiment(options, "upload_graph");
       LOG.info(
           "The job graph size ({} in bytes) is larger than {}. Automatically add "
               + "the upload_graph option to experiments.",
           jobGraphByteSize,
           CREATE_JOB_REQUEST_LIMIT_BYTES);
     }
 
-    if (hasExperiment(options, "upload_graph") && useUnifiedWorker(options)) {
+    if (useUnifiedWorker(options)) {
       ArrayList<String> experiments = new ArrayList<>(options.getExperiments());
       while (experiments.remove("upload_graph")) {}
       options.setExperiments(experiments);

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

@@ -55,6 +55,7 @@
 import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.WindmillStubFactoryFactory;
 import org.apache.beam.runners.dataflow.worker.windmill.work.WorkItemReceiver;
 import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.options.ExperimentalOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.util.BackOff;
 import org.apache.beam.sdk.util.BackOffUtils;
@@ -113,13 +114,13 @@ private static DataflowWorkerHarnessOptions testOptions(
     options.setProject("project");
     options.setJobId("job");
     options.setWorkerId("worker");
-    List<String> experiments =
-        options.getExperiments() == null ? new ArrayList<>() : options.getExperiments();
+
     if (enableStreamingEngine) {
-      experiments.add(GcpOptions.STREAMING_ENGINE_EXPERIMENT);
+      ExperimentalOptions.addExperiment(options, GcpOptions.STREAMING_ENGINE_EXPERIMENT);
+    }
+    for (String experiment : additionalExperiments) {
+      ExperimentalOptions.addExperiment(options, experiment);
     }
-    experiments.addAll(additionalExperiments);
-    options.setExperiments(experiments);
 
     options.setWindmillServiceStreamingRpcBatchLimit(Integer.MAX_VALUE);
     options.setWindmillServiceStreamingRpcHealthCheckPeriodMs(NO_HEALTH_CHECK);