[Task]: Enable to set timeout for Python TestPipeline

[Abacn]

What needs to happen?

Currently TestPipeline run indefinitely:

beam/sdks/python/apache_beam/testing/test_pipeline.py

Line 116 in aef2195

state = result.wait_until_finish()

In the case the test timeout, it does not print useful information, just a pytest timeout message and the stacktrace where it gets interrupted (e.g. https://github.com/apache/beam/runs/19275621816)

Failed: Timeout >4500.0s
self = <apache_beam.transforms.ptransform_test.PTransformTest testMethod=test_flatten_one_single_pcollection>

    @pytest.mark.it_validatesrunner
    def test_flatten_one_single_pcollection(self):
>     with TestPipeline() as pipeline:
...
        while thread.is_alive():
>         time.sleep(5.0)
E         Failed: Timeout >4500.0s

However, DataflowRunner.wait_until_finish() indeed supports duration:

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

Line 746 in aef2195

def wait_until_finish(self, duration=None):

and when timeout, it prints the job id so one can find the Dataflow job to investigate:

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

Lines 769 to 771 in aef2195

	assert duration or terminated, (
	'Job did not reach to a terminal state after waiting indefinitely. '
	'{}'.format(consoleUrl))

We should be able to use this functionality for TestPipeline, for example,

with TestPipeline(timeout = 600.0) as p:
   ...

Issue Priority

Priority: 3 (nice-to-have improvement)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[GautamGottipati]

.take-issue

[GautamGottipati]

@Abacn As far as I understand we have to add Timeout for TestPipeline. So I plan to pass this parameter while creation of TestPipeline class by initializing a varible self.duration = timeout in test_pipeline.__init__(duration=None) and pass this initialized variable to
state = result.wait_until_finish(duration=duration) . But I am unable to understand how is this wait_until_finish() function is called, it would be helpful if I get some explaination regarding this.
Secondly, Is my understanding correct.

[Tjindl]

.take-issue

[Tjindl]

Hello @Abacn, I looked around the Python code to get hints about the Java SDK component. I think someone has ticked the Python SDK component, but I suspect it is not solved. Could you please clarify whether the Python SDK component is actually solved or just a mistake?