[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?

[derrickaw]

.take-issue

[Naseer-010]

.take-issue

[Naseer-010]

hey @Abacn I think we just have to add timeout parameter as someone mentioned before and pass it to wait until finish. But the timeout time should be in seconds or milliseconds?
I have checked the code and in wait_until_finish() the duration is in milliseconds. Do you want me to use milliseconds as it is or it will be good to use seconds as user can also add fractional seconds so that he doesn't have to enter large numbers when they want to run for some minutes ?