Added Portable Runner Passing for Python Runners

Author: TongruiLi

sdks/python/apache_beam/runners/dataflow/dataflow_runner.py

@@ -591,6 +591,7 @@ def _add_runner_v2_missing_options(options):
   debug_options.add_experiment('use_unified_worker')
   debug_options.add_experiment('use_runner_v2')
   debug_options.add_experiment('use_portable_job_submission')
+  debug_options.add_experiment('enable_portable_runner')
 
 
 def _check_and_add_missing_options(options):
@@ -662,6 +663,8 @@ def _is_runner_v2_disabled(options):
   """Returns true if runner v2 is disabled."""
   debug_options = options.view_as(DebugOptions)
   return (
+      debug_options.lookup_experiment('disable_portable_runner') or
+      debug_options.lookup_experiment('enable_streaming_java_runner') or
       debug_options.lookup_experiment('disable_runner_v2') or
       debug_options.lookup_experiment('disable_runner_v2_until_2023') or
       debug_options.lookup_experiment('disable_runner_v2_until_v2.50') or

sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py

@@ -624,7 +624,8 @@ def test_batch_is_runner_v2(self):
     for expected in ['beam_fn_api',
                      'use_unified_worker',
                      'use_runner_v2',
-                     'use_portable_job_submission']:
+                     'use_portable_job_submission',
+                     'enable_portable_runner']:
       self.assertTrue(
           options.view_as(DebugOptions).lookup_experiment(expected, False),
           expected)
@@ -636,6 +637,7 @@ def test_streaming_is_runner_v2(self):
     for expected in ['beam_fn_api',
                      'use_unified_worker',
                      'use_runner_v2',
+                     'enable_portable_runner',
                      'use_portable_job_submission',
                      'enable_windmill_service',
                      'enable_streaming_engine']:
@@ -653,6 +655,7 @@ def test_dataflow_service_options_enable_prime_sets_runner_v2(self):
     for expected in ['beam_fn_api',
                      'use_unified_worker',
                      'use_runner_v2',
+                     'enable_portable_runner',
                      'use_portable_job_submission']:
       self.assertTrue(
           options.view_as(DebugOptions).lookup_experiment(expected, False),
@@ -669,6 +672,7 @@ def test_dataflow_service_options_enable_prime_sets_runner_v2(self):
                      'use_unified_worker',
                      'use_runner_v2',
                      'use_portable_job_submission',
+                     'enable_portable_runner',
                      'enable_windmill_service',
                      'enable_streaming_engine']:
       self.assertTrue(