Fix upload_graph condition regression and formatting issues

Author: ash6898

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

@@ -1244,7 +1244,8 @@ public DataflowPipelineJob run(Pipeline pipeline) {
     // 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")) {
+      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.");
@@ -1292,7 +1293,8 @@ public DataflowPipelineJob run(Pipeline pipeline) {
       {
         // Experiment marking that the harness supports tag encoding v2
         // Backend will enable tag encoding v2 only if the harness supports it.
-        ExperimentalOptions.addExperiment(options, "streaming_engine_state_tag_encoding_v2_supported");
+        ExperimentalOptions.addExperiment(
+            options, "streaming_engine_state_tag_encoding_v2_supported");
       }
 
       if (useUnifiedWorker(options)) {
@@ -1462,7 +1464,8 @@ public DataflowPipelineJob run(Pipeline pipeline) {
               .collect(Collectors.toList());
 
       if (minCpuFlags.isEmpty()) {
-        ExperimentalOptions.addExperiment(dataflowOptions, "min_cpu_platform=" + dataflowOptions.getMinCpuPlatform());
+        ExperimentalOptions.addExperiment(
+            dataflowOptions, "min_cpu_platform=" + dataflowOptions.getMinCpuPlatform());
       } else {
         LOG.warn(
             "Flag min_cpu_platform is defined in both top level PipelineOption, "
@@ -1498,8 +1501,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
     // enable upload_graph when the graph is too large
     byte[] jobGraphBytes = DataflowPipelineTranslator.jobToString(newJob).getBytes(UTF_8);
     int jobGraphByteSize = jobGraphBytes.length;
-    if (jobGraphByteSize >= CREATE_JOB_REQUEST_LIMIT_BYTES
-        && !useUnifiedWorker(options)) {
+    if (jobGraphByteSize >= CREATE_JOB_REQUEST_LIMIT_BYTES && !useUnifiedWorker(options)) {
       ExperimentalOptions.addExperiment(options, "upload_graph");
       LOG.info(
           "The job graph size ({} in bytes) is larger than {}. Automatically add "
@@ -1508,7 +1510,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
           CREATE_JOB_REQUEST_LIMIT_BYTES);
     }
 
-    if (useUnifiedWorker(options)) {
+    if (hasExperiment(options, "upload_graph") && useUnifiedWorker(options)) {
       ArrayList<String> experiments = new ArrayList<>(options.getExperiments());
       while (experiments.remove("upload_graph")) {}
       options.setExperiments(experiments);

sdks/python/apache_beam/io/external/xlang_parquetio_test.py

@@ -51,7 +51,8 @@ def test_xlang_parquetio_write(self):
     address = 'localhost:%s' % port
     try:
       with TestPipeline() as p:
-        p.get_pipeline_options().view_as(DebugOptions).add_experiment('jar_packages=' + expansion_jar)
+        p.get_pipeline_options().view_as(DebugOptions).add_experiment(
+            'jar_packages=' + expansion_jar)
         p.not_use_test_runner_api = True
         _ = p \
           | beam.Create([

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -151,7 +151,8 @@ def __init__(
       # add the flag if 'no_use_multiple_sdk_containers' is present.
       # TODO: Cleanup use_multiple_sdk_containers once we deprecate Python SDK
       # till version 2.4.
-      if ('no_use_multiple_sdk_containers' not in (self.debug_options.experiments or [])):
+      if ('no_use_multiple_sdk_containers'
+          not in (self.debug_options.experiments or [])):
         self.debug_options.add_experiment('use_multiple_sdk_containers')
     # FlexRS
     if self.google_cloud_options.flexrs_goal == 'COST_OPTIMIZED':