Apache Beam Integration causes ValueError with DoFn using Beam's userstate module

[criped]

How do you use Sentry?

Sentry Saas (sentry.io)

Version

Any (tried 1.5.12 and 1.9.5)

Steps to Reproduce

• python 3.8
• apache-beam==2.40.0
• pycharm
• DoFn using User State Specs start environment. Example:

import apache_beam as beam
from apache_beam.transforms.userstate import TimerSpec

class MyDoFn(beam.DoFn):
    TIMER_SPEC = TimerSpec('timer', beam.TimeDomain.REAL_TIME)

    def process(
        self,
        element: Tuple[str, str],
        w=beam.DoFn.WindowParam,
        stale_timer=beam.DoFn.TimerParam(TIMER_SPEC)
    ):

        key, value = element
        stale_timer.set(time.time() + 5 * 60)

    @on_timer(TIMER_SPEC)
    def my_timer_expiry_callback(self):
        print('Timer expired!')

• Run a Beam pipeline with the DoFn mentioned above. The error is raised regardless of the runner (tried DirectRunner and Dataflow)

Expected Result

The pipeline should go on without any error as it does when removing Apache Beam Integration for Sentry.

Actual Result

ERROR:apache_beam.runners.direct.executor:Exception at bundle <apache_beam.runners.direct.bundle_factory._Bundle object at 0x123147e40>, due to an exception.
 Traceback (most recent call last):
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 370, in call
    self.attempt_call(
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 404, in attempt_call
    evaluator.start_bundle()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 856, in start_bundle
    self.runner = DoFnRunner(
  File "apache_beam/runners/common.py", line 1366, in apache_beam.runners.common.DoFnRunner.__init__
  File "apache_beam/runners/common.py", line 290, in apache_beam.runners.common.DoFnSignature.__init__
  File "apache_beam/runners/common.py", line 319, in apache_beam.runners.common.DoFnSignature._validate
  File "apache_beam/runners/common.py", line 377, in apache_beam.runners.common.DoFnSignature._validate_stateful_dofn
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/transforms/userstate.py", line 305, in validate_stateful_dofn
    raise ValueError((
ValueError: The on_timer callback for TimerSpec(ts-timer) is not the specified .my_timer_expiry_callback method for DoFn <my_module.do_fns.MyDoFn object at 0x12311c520> (perhaps it was overwritten?).

ERROR:apache_beam.runners.direct.executor:Giving up after 4 attempts.
WARNING:apache_beam.runners.direct.executor:A task failed with exception: The on_timer callback for TimerSpec(ts-timer) is not the specified .my_timer_expiry_callback method for DoFn <my_module.do_fns.MyDoFn object at 0x12311c520> (perhaps it was overwritten?).
Traceback (most recent call last):
  File "/Users/criped/Code/test_project/data_pipeline.py", line 121, in <module>
    run()
  File "/Users/criped/Code/test_project/data_pipeline.py", line 114, in run
    p.run().wait_until_finish()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/direct_runner.py", line 588, in wait_until_finish
    self._executor.await_completion()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 432, in await_completion
    self._executor.await_completion()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 480, in await_completion
    raise update.exception
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 370, in call
    self.attempt_call(
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 404, in attempt_call
    evaluator.start_bundle()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 856, in start_bundle
    self.runner = DoFnRunner(
  File "apache_beam/runners/common.py", line 1366, in apache_beam.runners.common.DoFnRunner.__init__
  File "apache_beam/runners/common.py", line 290, in apache_beam.runners.common.DoFnSignature.__init__
  File "apache_beam/runners/common.py", line 319, in apache_beam.runners.common.DoFnSignature._validate
  File "apache_beam/runners/common.py", line 377, in apache_beam.runners.common.DoFnSignature._validate_stateful_dofn
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/transforms/userstate.py", line 305, in validate_stateful_dofn
    raise ValueError((
ValueError: The on_timer callback for TimerSpec(ts-timer) is not the specified .my_timer_expiry_callback method for DoFn <my_module.do_fns.MyDoFn object at 0x12311c520> (perhaps it was overwritten?).

Process finished with exit code 1

For further context, another DoFn with more State Specs (concretely a BagStateSpec, CombiningValueStateSpec, ReadModifyWriteStateSpec and TimerSpec) was tried and still a similar issue:

ERROR:apache_beam.runners.direct.executor:Exception at bundle <apache_beam.runners.direct.bundle_factory._Bundle object at 0x11a825080>, due to an exception.
 Traceback (most recent call last):
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 370, in call
    self.attempt_call(
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 404, in attempt_call
    evaluator.start_bundle()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 856, in start_bundle
    self.runner = DoFnRunner(
  File "apache_beam/runners/common.py", line 1366, in apache_beam.runners.common.DoFnRunner.__init__
  File "apache_beam/runners/common.py", line 290, in apache_beam.runners.common.DoFnSignature.__init__
  File "apache_beam/runners/common.py", line 319, in apache_beam.runners.common.DoFnSignature._validate
  File "apache_beam/runners/common.py", line 377, in apache_beam.runners.common.DoFnSignature._validate_stateful_dofn
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/transforms/userstate.py", line 288, in validate_stateful_dofn
    raise ValueError(
ValueError: DoFn <my_module.do_fns.FooBuffer object at 0x11a8723a0> has multiple StateSpecs with the same name: {BagStateSpec(rave), ReadModifyWriteStateSpec(size), ReadModifyWriteStateSpec(size), CombiningValueStateSpec(count), BagStateSpec(rave), CombiningValueStateSpec(count)}.

ERROR:apache_beam.runners.direct.executor:Giving up after 4 attempts.
WARNING:apache_beam.runners.direct.executor:A task failed with exception: DoFn <my_module.do_fns.FooBuffer object at 0x11a8723a0> has multiple StateSpecs with the same name: {BagStateSpec(rave), ReadModifyWriteStateSpec(size), ReadModifyWriteStateSpec(size), CombiningValueStateSpec(count), BagStateSpec(rave), CombiningValueStateSpec(count)}.
Traceback (most recent call last):
  File "/Users/criped/Code/test_project/data_pipeline.py", line 120, in <module>
    run()
  File "/Users/criped/Code/test_project/data_pipeline.py", line 113, in run
    p.run().wait_until_finish()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/direct_runner.py", line 588, in wait_until_finish
    self._executor.await_completion()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 432, in await_completion
    self._executor.await_completion()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 480, in await_completion
    raise update.exception
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 370, in call
    self.attempt_call(
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/executor.py", line 404, in attempt_call
    evaluator.start_bundle()
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 856, in start_bundle
    self.runner = DoFnRunner(
  File "apache_beam/runners/common.py", line 1366, in apache_beam.runners.common.DoFnRunner.__init__
  File "apache_beam/runners/common.py", line 290, in apache_beam.runners.common.DoFnSignature.__init__
  File "apache_beam/runners/common.py", line 319, in apache_beam.runners.common.DoFnSignature._validate
  File "apache_beam/runners/common.py", line 377, in apache_beam.runners.common.DoFnSignature._validate_stateful_dofn
  File "/Users/criped/Code/test_project/venv/lib/python3.8/site-packages/apache_beam/transforms/userstate.py", line 288, in validate_stateful_dofn
    raise ValueError(
ValueError: DoFn <my_module.do_fns.FooBuffer object at 0x11a8723a0> has multiple StateSpecs with the same name: {BagStateSpec(rave), ReadModifyWriteStateSpec(size), ReadModifyWriteStateSpec(size), CombiningValueStateSpec(count), BagStateSpec(rave), CombiningValueStateSpec(count)}.

Process finished with exit code 1

[szokeasaurusrex]

@criped, this issue has been open for awhile, so I would just like to confirm whether it is still a problem for you with the latest Sentry SDK version? If it is, could you please provide a full code sample that I can run to reproduce the problem (including the code that sets up and runs the pipeline)?

[getsantry[bot]]

This issue has gone three weeks without activity. In another week, I will close it.

But! If you comment or otherwise update it, I will reset the clock, and if you remove the label Waiting for: Community, I will leave it alone ... forever!

---

"A weed is but an unloved flower." ― Ella Wheeler Wilcox 🥀