[beam] branch release-2.7.1 updated: Default to PiplelineState.UNKNOWN when job state returned from v1beta3 cannot be recogninzed.

[altay]

This is an automated email from the ASF dual-hosted git repository.

altay pushed a commit to branch release-2.7.1
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/release-2.7.1 by this push:
     new 387b041  Default to PiplelineState.UNKNOWN when job state returned 
from v1beta3 cannot be recogninzed.
     new be007ad  Merge pull request #9137 from tvalentyn/cp_9094
387b041 is described below

commit 387b041e61d10d6b1f37b01171105f0d34fb178e
Author: Valentyn Tymofieiev <valen...@google.com>
AuthorDate: Wed Jul 17 17:06:53 2019 -0700

    Default to PiplelineState.UNKNOWN when job state returned from v1beta3 
cannot be recogninzed.
---
 sdks/python/apache_beam/runners/dataflow/dataflow_runner.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py 
b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
index 9f84305..9e42d82 100644
--- a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
@@ -1078,9 +1078,9 @@ class DataflowPipelineResult(PipelineResult):
   def _get_job_state(self):
     values_enum = dataflow_api.Job.CurrentStateValueValuesEnum
 
-    # TODO: Move this table to a another location.
-    # Ordered by the enum values.
-    api_jobstate_map = {
+    # Ordered by the enum values. Values that may be introduced in
+    # future versions of Dataflow API are considered UNKNOWN by the SDK.
+    api_jobstate_map = defaultdict(lambda: PipelineState.UNKNOWN, {
         values_enum.JOB_STATE_UNKNOWN: PipelineState.UNKNOWN,
         values_enum.JOB_STATE_STOPPED: PipelineState.STOPPED,
         values_enum.JOB_STATE_RUNNING: PipelineState.RUNNING,
@@ -1092,7 +1092,7 @@ class DataflowPipelineResult(PipelineResult):
         values_enum.JOB_STATE_DRAINED: PipelineState.DRAINED,
         values_enum.JOB_STATE_PENDING: PipelineState.PENDING,
         values_enum.JOB_STATE_CANCELLING: PipelineState.CANCELLING,
-    }
+    })
 
     return (api_jobstate_map[self._job.currentState] if self._job.currentState
             else PipelineState.UNKNOWN)