[Bug]: Windowing stops working and throws Cython error around IntervalWindowBase

[richardcann]

What happened?

Using Beam versions 2.37.0 and 2.43.0, we are reading from streaming source Pub/Sub and using two FixedWindows of 30 minutes each to aggregate elements and then group by a key we assigned based on the message. The pipeline works fine for about ~1hr (I can see windows are getting outputed correctly and process afterwards) and then it throws the following error around the IntervalWindowBase:

Error message from worker: generic::unknown: Traceback (most recent call last):
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 287, in _execute
    response = task()
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 360, in <lambda>
    lambda: self.create_worker().do_instruction(request), request)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 596, in do_instruction
    return getattr(self, request_type)(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 634, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1003, in process_bundle
    input_op_by_transform_id[element.transform_id].process_encoded(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 225, in process_encoded
    decoded_value = self.windowed_coder_impl.decode_from_stream(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1465, in decode_from_stream
    value = self._value_coder.decode_from_stream(in_stream, nested)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1008, in decode_from_stream
    return self._construct_from_components([
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1009, in <listcomp>
    c.decode_from_stream(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1194, in decode_from_stream
    elements = [
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1195, in <listcomp>
    self._elem_coder.decode_from_stream(in_stream, True)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1557, in decode_from_stream
    return self._value_coder.decode(in_stream.read(value_length))
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 240, in decode
    return self.decode_from_stream(create_InputStream(encoded), False)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 583, in decode_from_stream
    return self.fallback_coder_impl.decode_from_stream(stream, nested)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 273, in decode_from_stream
    return self._decoder(stream.read_all(nested))
AttributeError: Can't get attribute '__pyx_unpickle__IntervalWindowBase' on <module 'apache_beam.utils.windowed_value' from '/pipeline/.venv/lib/python3.9/site-packages/apache_beam/utils/windowed_value.py'>

generic::unknown: Traceback (most recent call last):
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 287, in _execute
    response = task()
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 360, in <lambda>
    lambda: self.create_worker().do_instruction(request), request)
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 596, in do_instruction
    return getattr(self, request_type)(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 634, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1003, in process_bundle
    input_op_by_transform_id[element.transform_id].process_encoded(
  File "/pipeline/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 225, in process_encoded
    decoded_value = self.windowed_coder_impl.decode_from_stream(
  File "apache_beam/coders/coder_impl.py", line 1446, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 1465, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 1009, in apache_beam.coders.coder_impl.AbstractComponentCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 1195, in apache_beam.coders.coder_impl.SequenceCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 1557, in apache_beam.coders.coder_impl.LengthPrefixCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 240, in apache_beam.coders.coder_impl.StreamCoderImpl.decode
  File "apache_beam/coders/coder_impl.py", line 583, in apache_beam.coders.coder_impl.FastPrimitivesCoderImpl.decode_from_stream
  File "apache_beam/coders/coder_impl.py", line 273, in apache_beam.coders.coder_impl.CallbackCoderImpl.decode_from_stream
  File "stringsource", line 17, in apache_beam.utils.windowed_value._IntervalWindowBase.__setstate_cython__
TypeError: Expected tuple, got dict

This error also happens when I reduce the size of the window. We are using Dataflow as the Runner. Input type to the windowing is Tuple[str, Dict[str, str]]

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• 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

[tvalentyn]

Thanks for reaching out.

1. Do you consistently reproduce the issue?
2. Is this a new pipeline that you didn't run before or the issue started after an SDK update?
3. Would you be able to provide a code snippet that reproduces the issue?
4. Could you add more details about:

Input type to the windowing is Tuple[str, Dict[str, str]]

[tvalentyn]

Friendly reminder about the questions above. Thank you!

[tvalentyn]

Closing for now due to lack of response - feel free to reopen if this is still an issue.