Bump dataflow client version, restore pausing/paused concept

Author: jrmccluskey

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

@@ -759,8 +759,8 @@ def api_jobstate_to_pipeline_state(api_jobstate):
             values_enum.JOB_STATE_CANCELLING: PipelineState.CANCELLING,
             values_enum.JOB_STATE_RESOURCE_CLEANING_UP: PipelineState.
             RESOURCE_CLEANING_UP,
-            # values_enum.JOB_STATE_PAUSING : PipelineState.PAUSING,
-            # values_enum.JOB_STATE_PAUSED : PipelineState.PAUSED,
+            values_enum.JOB_STATE_PAUSING : PipelineState.PAUSING,
+            values_enum.JOB_STATE_PAUSED : PipelineState.PAUSED,
         })
 
     return (

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

@@ -197,6 +197,22 @@ def get_job_side_effect(*args, **kwargs):
       result = duration_timedout_result.wait_until_finish(5000)
       self.assertEqual(result, PipelineState.RUNNING)
 
+    with mock.patch('time.time', mock.MagicMock(side_effect=[1, 9, 9, 20, 20])):
+      duration_timedout_runner = MockDataflowRunner(
+          [values_enum.JOB_STATE_PAUSING])
+      duration_timedout_result = DataflowPipelineResult(
+          duration_timedout_runner.job, duration_timedout_runner, options)
+      result = duration_timedout_result.wait_until_finish(5000)
+      self.assertEqual(result, PipelineState.PAUSING)
+
+    with mock.patch('time.time', mock.MagicMock(side_effect=[1, 9, 9, 20, 20])):
+      duration_timedout_runner = MockDataflowRunner(
+          [values_enum.JOB_STATE_PAUSED])
+      duration_timedout_result = DataflowPipelineResult(
+          duration_timedout_runner.job, duration_timedout_runner, options)
+      result = duration_timedout_result.wait_until_finish(5000)
+      self.assertEqual(result, PipelineState.PAUSED)
+
     with mock.patch('time.time', mock.MagicMock(side_effect=[1, 1, 2, 2, 3])):
       with self.assertRaisesRegex(DataflowRuntimeException,
                                   'Dataflow pipeline failed. State: CANCELLED'):
@@ -255,6 +271,8 @@ def test_api_jobstate_to_pipeline_state(self):
         (values_enum.JOB_STATE_DRAINED, PipelineState.DRAINED),
         (values_enum.JOB_STATE_PENDING, PipelineState.PENDING),
         (values_enum.JOB_STATE_CANCELLING, PipelineState.CANCELLING),
+        (values_enum.JOB_STATE_PAUSING, PipelineState.PAUSING),
+        (values_enum.JOB_STATE_PAUSED, PipelineState.PAUSED),
         (
             values_enum.JOB_STATE_RESOURCE_CLEANING_UP,
             PipelineState.RESOURCE_CLEANING_UP),

sdks/python/setup.py

@@ -486,7 +486,7 @@ def get_portability_package_data():
               'google-cloud-datastore>=2.0.0,<3',
               'google-cloud-pubsub>=2.1.0,<3',
               'google-cloud-storage>=2.18.2,<3',
-              'google-cloud-dataflow-client>=0.11.0,<0.12.0',
+              'google-cloud-dataflow-client>=0.13.0,<0.14.0',
               # GCP packages required by tests
               'google-cloud-bigquery>=2.0.0,<4',
               'google-cloud-bigquery-storage>=2.6.3,<3',