You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2023/05/08 16:59:44 UTC
[beam] branch master updated: [Python SDK] wait_until_finish always raises the exception for FAILED jobs (#26206)
This is an automated email from the ASF dual-hosted git repository.
kenn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 26e6dec4ced [Python SDK] wait_until_finish always raises the exception for FAILED jobs (#26206)
26e6dec4ced is described below
commit 26e6dec4cede23393529359478025c2a273d3c8d
Author: liferoad <hu...@gmail.com>
AuthorDate: Mon May 8 12:59:33 2023 -0400
[Python SDK] wait_until_finish always raises the exception for FAILED jobs (#26206)
Authored-by: xqhu <xq...@google.com>
---
.../runners/dataflow/dataflow_runner.py | 8 +++--
.../runners/dataflow/dataflow_runner_test.py | 39 ++++++++++++----------
2 files changed, 28 insertions(+), 19 deletions(-)
diff --git a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
index e98e931350a..63bfc07cdc5 100644
--- a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
@@ -1556,8 +1556,6 @@ class DataflowPipelineResult(PipelineResult):
'Job did not reach to a terminal state after waiting indefinitely. '
'{}'.format(consoleUrl))
- # TODO(https://github.com/apache/beam/issues/21695): Also run this check
- # if wait_until_finish was called after the pipeline completed.
if terminated and self.state != PipelineState.DONE:
# TODO(BEAM-1290): Consider converting this to an error log based on
# theresolution of the issue.
@@ -1566,6 +1564,12 @@ class DataflowPipelineResult(PipelineResult):
'Dataflow pipeline failed. State: %s, Error:\n%s' %
(self.state, getattr(self._runner, 'last_error_msg', None)),
self)
+ elif PipelineState.is_terminal(
+ self.state) and self.state == PipelineState.FAILED and self._runner:
+ raise DataflowRuntimeException(
+ 'Dataflow pipeline failed. State: %s, Error:\n%s' %
+ (self.state, getattr(self._runner, 'last_error_msg', None)),
+ self)
return self.state
diff --git a/sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py b/sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py
index aca1def5942..b00644d13fd 100644
--- a/sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py
@@ -137,6 +137,11 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
failed_result = DataflowPipelineResult(failed_runner.job, failed_runner)
failed_result.wait_until_finish()
+ # check the second call can still triggers the exception
+ with self.assertRaisesRegex(DataflowRuntimeException,
+ 'Dataflow pipeline failed. State: FAILED'):
+ failed_result.wait_until_finish()
+
succeeded_runner = MockDataflowRunner([values_enum.JOB_STATE_DONE])
succeeded_result = DataflowPipelineResult(
succeeded_runner.job, succeeded_runner)
@@ -511,13 +516,13 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
job_dict = json.loads(str(runner.job))
self.assertIn(
- u'CombineValues', set(step[u'kind'] for step in job_dict[u'steps']))
+ 'CombineValues', set(step['kind'] for step in job_dict['steps']))
def _find_step(self, job, step_name):
job_dict = json.loads(str(job))
maybe_step = [
- s for s in job_dict[u'steps']
- if s[u'properties'][u'user_name'] == step_name
+ s for s in job_dict['steps']
+ if s['properties']['user_name'] == step_name
]
self.assertTrue(maybe_step, 'Could not find step {}'.format(step_name))
return maybe_step[0]
@@ -542,12 +547,12 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
}]
step = self._find_step(job, step_name)
- self.assertEqual(step[u'kind'], step_kind)
+ self.assertEqual(step['kind'], step_kind)
# The display data here is forwarded because the replace transform is
# subclassed from iobase.Read.
- self.assertGreater(len(step[u'properties']['display_data']), 0)
- self.assertEqual(step[u'properties']['output_info'], expected_output_info)
+ self.assertGreater(len(step['properties']['display_data']), 0)
+ self.assertEqual(step['properties']['output_info'], expected_output_info)
def test_read_create_translation(self):
runner = DataflowRunner()
@@ -557,7 +562,7 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
# pylint: disable=expression-not-assigned
p | beam.Create([b'a', b'b', b'c'])
- self.expect_correct_override(runner.job, u'Create/Read', u'ParallelRead')
+ self.expect_correct_override(runner.job, 'Create/Read', 'ParallelRead')
def test_read_pubsub_translation(self):
runner = DataflowRunner()
@@ -570,7 +575,7 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
p | beam.io.ReadFromPubSub(topic='projects/project/topics/topic')
self.expect_correct_override(
- runner.job, u'ReadFromPubSub/Read', u'ParallelRead')
+ runner.job, 'ReadFromPubSub/Read', 'ParallelRead')
def test_gbk_translation(self):
runner = DataflowRunner()
@@ -604,10 +609,10 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
"user_name": "GroupByKey.out"
}] # yapf: disable
- gbk_step = self._find_step(runner.job, u'GroupByKey')
- self.assertEqual(gbk_step[u'kind'], u'GroupByKey')
+ gbk_step = self._find_step(runner.job, 'GroupByKey')
+ self.assertEqual(gbk_step['kind'], 'GroupByKey')
self.assertEqual(
- gbk_step[u'properties']['output_info'], expected_output_info)
+ gbk_step['properties']['output_info'], expected_output_info)
@unittest.skip(
'https://github.com/apache/beam/issues/18716: enable once '
@@ -657,24 +662,24 @@ class DataflowRunnerTest(unittest.TestCase, ExtraAssertionsMixin):
input | beam.GroupIntoBatches.WithShardedKey(2)
| beam.Map(lambda key_values: (key_values[0].key, key_values[1])))
step_name = (
- u'WithShardedKey/GroupIntoBatches/ParDo(_GroupIntoBatchesDoFn)')
+ 'WithShardedKey/GroupIntoBatches/ParDo(_GroupIntoBatchesDoFn)')
else:
input | beam.GroupIntoBatches(2)
- step_name = u'GroupIntoBatches/ParDo(_GroupIntoBatchesDoFn)'
+ step_name = 'GroupIntoBatches/ParDo(_GroupIntoBatchesDoFn)'
return self._find_step(runner.job, step_name)['properties']
def test_group_into_batches_translation(self):
properties = self._run_group_into_batches_and_get_step_properties(
True, ['--enable_streaming_engine', '--experiments=use_runner_v2'])
- self.assertEqual(properties[PropertyNames.USES_KEYED_STATE], u'true')
- self.assertEqual(properties[PropertyNames.ALLOWS_SHARDABLE_STATE], u'true')
- self.assertEqual(properties[PropertyNames.PRESERVES_KEYS], u'true')
+ self.assertEqual(properties[PropertyNames.USES_KEYED_STATE], 'true')
+ self.assertEqual(properties[PropertyNames.ALLOWS_SHARDABLE_STATE], 'true')
+ self.assertEqual(properties[PropertyNames.PRESERVES_KEYS], 'true')
def test_group_into_batches_translation_non_sharded(self):
properties = self._run_group_into_batches_and_get_step_properties(
False, ['--enable_streaming_engine', '--experiments=use_runner_v2'])
- self.assertEqual(properties[PropertyNames.USES_KEYED_STATE], u'true')
+ self.assertEqual(properties[PropertyNames.USES_KEYED_STATE], 'true')
self.assertNotIn(PropertyNames.ALLOWS_SHARDABLE_STATE, properties)
self.assertNotIn(PropertyNames.PRESERVES_KEYS, properties)