[
https://issues.apache.org/jira/browse/BEAM-9767?focusedWorklogId=423666&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-423666
]
ASF GitHub Bot logged work on BEAM-9767:
----------------------------------------
Author: ASF GitHub Bot
Created on: 16/Apr/20 19:13
Start Date: 16/Apr/20 19:13
Worklog Time Spent: 10m
Work Description: pabloem commented on issue #11440: [BEAM-9767] Add a
timeout to the TestStream GRPC and fix the Streaming cache timeout
URL: https://github.com/apache/beam/pull/11440#issuecomment-614844071
retest this please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 423666)
Time Spent: 0.5h (was: 20m)
> test_streaming_wordcount flaky timeouts
> ---------------------------------------
>
> Key: BEAM-9767
> URL: https://issues.apache.org/jira/browse/BEAM-9767
> Project: Beam
> Issue Type: Bug
> Components: sdk-py-core, test-failures
> Reporter: Udi Meiri
> Assignee: Sam Rohde
> Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> Timed out after 600s, typically completes in 2.8s on my workstation.
> https://builds.apache.org/job/beam_PreCommit_Python_Commit/12376/
> {code}
> self =
> <apache_beam.runners.interactive.interactive_runner_test.InteractiveRunnerTest
> testMethod=test_streaming_wordcount>
> @unittest.skipIf(
> sys.version_info < (3, 5, 3),
> 'The tests require at least Python 3.6 to work.')
> def test_streaming_wordcount(self):
> class WordExtractingDoFn(beam.DoFn):
> def process(self, element):
> text_line = element.strip()
> words = text_line.split()
> return words
>
> # Add the TestStream so that it can be cached.
> ib.options.capturable_sources.add(TestStream)
> ib.options.capture_duration = timedelta(seconds=5)
>
> p = beam.Pipeline(
> runner=interactive_runner.InteractiveRunner(),
> options=StandardOptions(streaming=True))
>
> data = (
> p
> | TestStream()
> .advance_watermark_to(0)
> .advance_processing_time(1)
> .add_elements(['to', 'be', 'or', 'not', 'to', 'be'])
> .advance_watermark_to(20)
> .advance_processing_time(1)
> .add_elements(['that', 'is', 'the', 'question'])
> | beam.WindowInto(beam.window.FixedWindows(10))) # yapf: disable
>
> counts = (
> data
> | 'split' >> beam.ParDo(WordExtractingDoFn())
> | 'pair_with_one' >> beam.Map(lambda x: (x, 1))
> | 'group' >> beam.GroupByKey()
> | 'count' >> beam.Map(lambda wordones: (wordones[0],
> sum(wordones[1]))))
>
> # Watch the local scope for Interactive Beam so that referenced
> PCollections
> # will be cached.
> ib.watch(locals())
>
> # This is normally done in the interactive_utils when a transform is
> # applied but needs an IPython environment. So we manually run this
> here.
> ie.current_env().track_user_pipelines()
>
> # Create a fake limiter that cancels the BCJ once the main job receives
> the
> # expected amount of results.
> class FakeLimiter:
> def __init__(self, p, pcoll):
> self.p = p
> self.pcoll = pcoll
>
> def is_triggered(self):
> result = ie.current_env().pipeline_result(self.p)
> if result:
> try:
> results = result.get(self.pcoll)
> except ValueError:
> return False
> return len(results) >= 10
> return False
>
> # This sets the limiters to stop reading when the test receives 10
> elements
> # or after 5 seconds have elapsed (to eliminate the possibility of
> hanging).
> ie.current_env().options.capture_control.set_limiters_for_test(
> [FakeLimiter(p, data), DurationLimiter(timedelta(seconds=5))])
>
> # This tests that the data was correctly cached.
> pane_info = PaneInfo(True, True, PaneInfoTiming.UNKNOWN, 0, 0)
> expected_data_df = pd.DataFrame([
> ('to', 0, [IntervalWindow(0, 10)], pane_info),
> ('be', 0, [IntervalWindow(0, 10)], pane_info),
> ('or', 0, [IntervalWindow(0, 10)], pane_info),
> ('not', 0, [IntervalWindow(0, 10)], pane_info),
> ('to', 0, [IntervalWindow(0, 10)], pane_info),
> ('be', 0, [IntervalWindow(0, 10)], pane_info),
> ('that', 20000000, [IntervalWindow(20, 30)], pane_info),
> ('is', 20000000, [IntervalWindow(20, 30)], pane_info),
> ('the', 20000000, [IntervalWindow(20, 30)], pane_info),
> ('question', 20000000, [IntervalWindow(20, 30)], pane_info)
> ], columns=[0, 'event_time', 'windows', 'pane_info']) # yapf: disable
>
> > data_df = ib.collect(data, include_window_info=True)
> apache_beam/runners/interactive/interactive_runner_test.py:237:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> apache_beam/runners/interactive/interactive_beam.py:451: in collect
> return head(pcoll, n=-1, include_window_info=include_window_info)
> apache_beam/runners/interactive/utils.py:204: in run_within_progress_indicator
> return func(*args, **kwargs)
> apache_beam/runners/interactive/interactive_beam.py:515: in head
> result.wait_until_finish()
> apache_beam/runners/interactive/interactive_runner.py:250: in
> wait_until_finish
> self._underlying_result.wait_until_finish()
> apache_beam/runners/direct/direct_runner.py:455: in wait_until_finish
> self._executor.await_completion()
> apache_beam/runners/direct/executor.py:439: in await_completion
> self._executor.await_completion()
> apache_beam/runners/direct/executor.py:484: in await_completion
> update = self.visible_updates.take()
> apache_beam/runners/direct/executor.py:557: in take
> item = self._queue.get(timeout=1)
> /usr/lib/python3.6/queue.py:173: in get
> self.not_empty.wait(remaining)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> self = <Condition(<unlocked _thread.lock object at 0x7f2f85244198>, 0)>
> timeout = 0.9999979329295456
> def wait(self, timeout=None):
> """Wait until notified or until a timeout occurs.
>
> If the calling thread has not acquired the lock when this method is
> called, a RuntimeError is raised.
>
> This method releases the underlying lock, and then blocks until it is
> awakened by a notify() or notify_all() call for the same condition
> variable in another thread, or until the optional timeout occurs. Once
> awakened or timed out, it re-acquires the lock and returns.
>
> When the timeout argument is present and not None, it should be a
> floating point number specifying a timeout for the operation in
> seconds
> (or fractions thereof).
>
> When the underlying lock is an RLock, it is not released using its
> release() method, since this may not actually unlock the lock when it
> was acquired multiple times recursively. Instead, an internal
> interface
> of the RLock class is used, which really unlocks it even when it has
> been recursively acquired several times. Another internal interface is
> then used to restore the recursion level when the lock is reacquired.
>
> """
> if not self._is_owned():
> raise RuntimeError("cannot wait on un-acquired lock")
> waiter = _allocate_lock()
> waiter.acquire()
> self._waiters.append(waiter)
> saved_state = self._release_save()
> gotit = False
> try: # restore state no matter what (e.g., KeyboardInterrupt)
> if timeout is None:
> waiter.acquire()
> gotit = True
> else:
> if timeout > 0:
> > gotit = waiter.acquire(True, timeout)
> E Failed: Timeout >600.0s
> /usr/lib/python3.6/threading.py:299: Failed
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
