Updated All Externally Visible Logs/Docs from Runner V2 to Portable Runner (#38532)

* Aded portable runner to python and go runners

* Removed empty if, added disable unit test for python.

* Removed unnessary class in unit test

* Removed unused variable

* Ran formatter

* Updated All Reference of Runner V2/V1 to portable runner/streaming java runner

* Specified portable runner as dataflow portable runner

* Updated testing logic to make things more clear

* spotless

* python spotless

* fixed broken test

* removed unnessary regex

Author: TongruiLi

runners/google-cloud-dataflow-java/arm/build.gradle

@@ -118,7 +118,7 @@ task printrunnerV2PipelineOptionsARM {
   dependsOn buildAndPushDockerJavaMultiarchContainer
 
   doLast {
-    println "To run a Dataflow job with runner V2 on ARM, add the following pipeline options to your command-line:"
+    println "To run a Dataflow job with Dataflow Portable Runner on ARM, add the following pipeline options to your command-line:"
     println runnerV2PipelineOptionsARM.join(' ')
   }
 }

runners/google-cloud-dataflow-java/build.gradle

@@ -431,7 +431,7 @@ task printRunnerV2PipelineOptions {
   dependsOn buildAndPushDockerJavaContainer
 
   doLast {
-    println "To run a Dataflow job with runner V2, add the following pipeline options to your command-line:"
+    println "To run a Dataflow job with Dataflow Portable Runner, add the following pipeline options to your command-line:"
     println runnerV2PipelineOptions.join(' ')
     println "Please delete your image upon completion with the following command:"
     println "docker rmi ${dockerJavaImageName}; gcloud container images delete --force-delete-tags ${dockerJavaImageName}"
@@ -471,7 +471,7 @@ def validatesRunnerStreamingConfig = [
   excludedTests: [
     // TODO(https://github.com/apache/beam/issues/21472)
     'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testAfterProcessingTimeContinuationTriggerUsingState',
-    // GroupIntoBatches.withShardedKey not supported on streaming runner v1
+    // GroupIntoBatches.withShardedKey not supported on Streaming Java Runner
     // https://github.com/apache/beam/issues/22592
     'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testWithShardedKeyInGlobalWindow',
 
@@ -570,7 +570,7 @@ createCrossLanguageValidatesRunnerTask(
 
 task validatesRunnerV2 {
   group = "Verification"
-  description = "Runs the ValidatesRunner tests on Dataflow Runner V2"
+  description = "Runs the ValidatesRunner tests on Dataflow Portable Runner"
   dependsOn(createRunnerV2ValidatesRunnerTest(
     name: 'validatesRunnerV2Test',
     pipelineOptions: runnerV2PipelineOptions,
@@ -610,7 +610,7 @@ task validatesRunnerV2 {
 
 task validatesRunnerV2Streaming {
   group = "Verification"
-  description = "Runs the ValidatesRunner tests on Dataflow Runner V2 forcing streaming mode"
+  description = "Runs the ValidatesRunner tests on Dataflow Portable Runner forcing streaming mode"
   dependsOn(createRunnerV2ValidatesRunnerTest(
     name: 'validatesRunnerV2TestStreaming',
     pipelineOptions: runnerV2PipelineOptions + ['--streaming', '--experiments=enable_streaming_engine'],
@@ -880,7 +880,7 @@ task postCommit {
 
 task postCommitRunnerV2 {
   group = "Verification"
-  description = "Various integration tests using the Dataflow runner V2."
+  description = "Various integration tests using the Dataflow Portable Runner."
   dependsOn googleCloudPlatformRunnerV2IntegrationTest
   dependsOn coreSDKJavaRunnerV2IntegrationTest
 }

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

@@ -1242,12 +1242,12 @@ private static boolean includesTransformUpgrades(Pipeline pipeline) {
   @Override
   public DataflowPipelineJob run(Pipeline pipeline) {
     // Multi-language pipelines and pipelines that include upgrades should automatically be upgraded
-    // to Runner v2.
+    // to Dataflow Portable Runner.
     if (DataflowRunner.isMultiLanguagePipeline(pipeline) || includesTransformUpgrades(pipeline)) {
       if (!useUnifiedWorker(options)) {
         List<String> experiments = firstNonNull(options.getExperiments(), Collections.emptyList());
         LOG.info(
-            "Automatically enabling Dataflow Runner v2 since the pipeline used cross-language"
+            "Automatically enabling Dataflow Portable Runner 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());
@@ -1260,7 +1260,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
           || hasExperiment(options, "disable_portable_runner")
           || hasExperiment(options, "enable_streaming_java_runner")) {
         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.");
+            "Dataflow Portable Runner both disabled and enabled: at least one of ['enable_portable_runner', 'beam_fn_api', 'use_unified_worker', 'use_runner_v2', 'use_portable_job_submission'] is set and also one of ['enable_streaming_java_runner', 'disable_portable_runner', '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
@@ -1374,10 +1374,13 @@ public DataflowPipelineJob run(Pipeline pipeline) {
     options.as(SdkHarnessOptions.class).setPipelineProtoHash(pipelineProtoHash);
 
     if (useUnifiedWorker(options)) {
-      LOG.info("Skipping v1 transform replacements since job will run on v2.");
+      LOG.info(
+          "Skipping Dataflow Streaming Java Runner transform replacements since job will run on Dataflow Portable Runner.");
     } else {
-      // Now rewrite things to be as needed for v1 (mutates the pipeline)
-      // This way the job submitted is valid for v1 and v2, simultaneously
+      // Now rewrite things to be as needed for Dataflow Streaming Java Runner (mutates the
+      // pipeline)
+      // This way the job submitted is valid for Dataflow Streaming Java Runner and Dataflow
+      // Portable Runner, simultaneously
       replaceV1Transforms(pipeline);
     }
     // Capture the SdkComponents for look up during step translations
@@ -1388,7 +1391,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
             .addAllDependencies(getDefaultArtifacts())
             .addAllCapabilities(Environments.getJavaCapabilities())
             .build());
-    // No need to perform transform upgrading for the Runner v1 proto.
+    // No need to perform transform upgrading for the Dataflow Streaming Java Runner proto.
     RunnerApi.Pipeline dataflowV1PipelineProto =
         PipelineTranslation.toProto(pipeline, dataflowV1Components, true, false);
 
@@ -1544,7 +1547,7 @@ public DataflowPipelineJob run(Pipeline pipeline) {
       options.setExperiments(experiments);
       LOG.warn(
           "The upload_graph experiment was specified, but it does not apply "
-              + "to runner v2 jobs. Option has been automatically removed.");
+              + "to Dataflow Portable Runner jobs. Option has been automatically removed.");
     }
 
     // Upload the job to GCS and remove the graph object from the API call.  The graph

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

@@ -1855,7 +1855,10 @@ public void testSettingConflictingEnableAndDisableExperimentsThrowsException() t
         ExperimentalOptions.addExperiment(options, disabledExperiment);
         Pipeline p = Pipeline.create(options);
         p.apply(Create.of("A"));
-        assertThrows("Runner V2 both disabled and enabled", IllegalArgumentException.class, p::run);
+        assertThrows(
+            "Dataflow Portable Runner both disabled and enabled",
+            IllegalArgumentException.class,
+            p::run);
       }
     }
   }

runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto

@@ -979,7 +979,7 @@ enum ConnectivityType {
   CONNECTIVITY_TYPE_DIRECTPATH = 2;
 }
 
-// Settings to control runtime behavior of the java runner v1 user worker.
+// Settings to control runtime behavior of the Streaming Java Runner user worker.
 message UserWorkerRunnerV1Settings {
   optional UserWorkerGrpcFlowControlSettings flow_control_settings = 3;
 

sdks/go/pkg/beam/runners/dataflow/dataflow.go

@@ -334,7 +334,7 @@ func getJobOptions(ctx context.Context, streaming bool) (*dataflowlib.JobOptions
 
 	experiments := jobopts.GetExperiments()
 	// Ensure that we enable the same set of experiments across all SDKs
-	// for runner v2.
+	// for Dataflow Portable Runner.
 	var fnApiSet, v2set, uwSet, portaSubmission, seSet, wsSet bool
 	for _, e := range experiments {
 		if strings.Contains(e, "beam_fn_api") {
@@ -349,8 +349,9 @@ func getJobOptions(ctx context.Context, streaming bool) (*dataflowlib.JobOptions
 		if strings.Contains(e, "use_portable_job_submission") {
 			portaSubmission = true
 		}
-		if strings.Contains(e, "disable_runner_v2") || strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e, "disable_prime_runner_v2") {
-			return nil, errors.New("detected one of the following experiments: disable_runner_v2 | disable_runner_v2_until_2023 | disable_prime_runner_v2. Disabling runner v2 is no longer supported as of Beam version 2.45.0+")
+		// enable_portable_runner is not documented and hence wont be set by default. This will be fixed in later versions.
+		if strings.Contains(e, "disable_runner_v2") || strings.Contains(e, "disable_runner_v2_until_2023") || strings.Contains(e, "disable_prime_runner_v2") || strings.Contains(e, "disable_portable_runner") || strings.Contains(e, "enable_streaming_java_runner") {
+			return nil, errors.New("detected one of the following experiments: disable_runner_v2 | disable_runner_v2_until_2023 | disable_prime_runner_v2 | disable_portable_runner | enable_streaming_java_runner. Disabling Dataflow Portable Runner is no longer supported as of Beam version 2.45.0+")
 		}
 	}
 	// Enable default experiments.
@@ -368,7 +369,7 @@ func getJobOptions(ctx context.Context, streaming bool) (*dataflowlib.JobOptions
 	}
 
 	// Ensure that streaming specific experiments are set for streaming pipelines
-	// since runner v2 only supports using streaming engine.
+	// since Dataflow Portable Runner only supports using streaming engine.
 	if streaming {
 		if !seSet {
 			experiments = append(experiments, "enable_streaming_engine")

sdks/go/pkg/beam/runners/dataflow/dataflow_test.go

@@ -20,6 +20,7 @@ import (
 	"flag"
 	"reflect"
 	"sort"
+	"strings"
 	"testing"
 
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core"
@@ -227,7 +228,7 @@ func TestGetJobOptions_NoExperimentsSetStreaming(t *testing.T) {
 	}
 }
 
-func TestGetJobOptions_DisableRunnerV2ExperimentsSet(t *testing.T) {
+func TestGetJobOptions_DisableRunnerV2ExperimentsSetFailJob(t *testing.T) {
 	resetGlobals()
 	*stagingLocation = "gs://testStagingLocation"
 	*gcpopts.Project = "testProject"
@@ -238,6 +239,46 @@ func TestGetJobOptions_DisableRunnerV2ExperimentsSet(t *testing.T) {
 
 	if err == nil {
 		t.Error("getJobOptions() returned error nil, want an error")
+	} else if !strings.Contains(err.Error(), "Disabling Dataflow Portable Runner is no longer supported") {
+		t.Errorf("getJobOptions() returned wrong error %q, want it to mention %q", err.Error(), "Disabling Dataflow Portable Runner is no longer supported")
+	}
+	if opts != nil {
+		t.Errorf("getJobOptions() returned JobOptions when it should not have, got %#v, want nil", opts)
+	}
+}
+
+func TestGetJobOptions_DisablePortableRunnerExperimentsSetFailJob(t *testing.T) {
+	resetGlobals()
+	*stagingLocation = "gs://testStagingLocation"
+	*gcpopts.Project = "testProject"
+	*gcpopts.Region = "testRegion"
+	*jobopts.Experiments = "disable_portable_runner"
+
+	opts, err := getJobOptions(context.Background(), false)
+
+	if err == nil {
+		t.Error("getJobOptions() returned error nil, want an error")
+	} else if !strings.Contains(err.Error(), "Disabling Dataflow Portable Runner is no longer supported") {
+		t.Errorf("getJobOptions() returned wrong error %q, want it to mention %q", err.Error(), "Disabling Dataflow Portable Runner is no longer supported")
+	}
+	if opts != nil {
+		t.Errorf("getJobOptions() returned JobOptions when it should not have, got %#v, want nil", opts)
+	}
+}
+
+func TestGetJobOptions_EnableStreamingJavaRunnerExperimentsSetFailJob(t *testing.T) {
+	resetGlobals()
+	*stagingLocation = "gs://testStagingLocation"
+	*gcpopts.Project = "testProject"
+	*gcpopts.Region = "testRegion"
+	*jobopts.Experiments = "enable_streaming_java_runner"
+
+	opts, err := getJobOptions(context.Background(), false)
+
+	if err == nil {
+		t.Error("getJobOptions() returned error nil, want an error")
+	} else if !strings.Contains(err.Error(), "Disabling Dataflow Portable Runner is no longer supported") {
+		t.Errorf("getJobOptions() returned wrong error %q, want it to mention %q", err.Error(), "Disabling Dataflow Portable Runner is no longer supported")
 	}
 	if opts != nil {
 		t.Errorf("getJobOptions() returned JobOptions when it should not have, got %#v, want nil", opts)

sdks/python/apache_beam/examples/cookbook/ordered_window_elements/streaming.py

@@ -135,7 +135,7 @@ def _state_read_range(self, buffer_state, range_lo, range_hi):
 
   def _state_clear_range(self, buffer_state, range_lo, range_hi):
     """Clears a specified range of elements from the buffer state."""
-    # TODO: Dataflow runner v2 gets stuck when MIN_TIMESTAMP is used
+    # TODO: Dataflow Portable Runner gets stuck when MIN_TIMESTAMP is used
     # as the lower bound for clear_range. Investigate this further.
     buffer_state.clear_range(range_lo, range_hi)
 

sdks/python/apache_beam/examples/kafkataxi/README.md

@@ -67,7 +67,7 @@ Perform Beam runner specific setup.
 
 ℹ️ Note that cross-language transforms require
 portable implementations of Spark/Flink/Direct runners. Dataflow requires
-[runner V2](https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#dataflow-runner-v2).
+[Dataflow Portable Runner](https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#dataflow-runner-v2).
 See [here](https://beam.apache.org/documentation/runners/dataflow/) for
 instructions for setting up Dataflow.
 

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

@@ -563,7 +563,7 @@ def test_streaming_with_fixed_num_streams(self):
 
   @unittest.skip(
       "Streaming to the Storage Write API sink with autosharding is broken "
-      "with Dataflow Runner V2.")
+      "with Dataflow Portable Runner.")
   def test_streaming_with_auto_sharding(self):
     self.skip_if_not_dataflow_runner()
     table = 'streaming_with_auto_sharding'