You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@beam.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2020/05/06 21:13:00 UTC
[jira] [Work logged] (BEAM-9767) test_streaming_wordcount flaky
timeouts
[ https://issues.apache.org/jira/browse/BEAM-9767?focusedWorklogId=431434&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-431434 ]
ASF GitHub Bot logged work on BEAM-9767:
----------------------------------------
Author: ASF GitHub Bot
Created on: 06/May/20 21:12
Start Date: 06/May/20 21:12
Worklog Time Spent: 10m
Work Description: rohdesamuel commented on pull request #11624:
URL: https://github.com/apache/beam/pull/11624#issuecomment-624893653
R: @pabloem
----------------------------------------------------------------
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:
users@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 431434)
Time Spent: 1h 20m (was: 1h 10m)
> 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: Critical
> Time Spent: 1h 20m
> 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)