bvolpato commented on code in PR #40507:
URL: https://github.com/apache/beam/pull/40507#discussion_r4425834411
##########
sdks/python/apache_beam/transforms/async_dofn.py:
##########
@@ -209,9 +212,25 @@ def setup(self):
AsyncWrapper._items_in_buffer[self._uuid] = 0
def teardown(self):
- """Forwards to the wrapped dofn's teardown method."""
+ """Waits for this instance's work before tearing down the wrapped dofn."""
+ if self._use_asyncio:
Review Comment:
I checked the runner lifecycle behind this case. In the [portable
runner](https://github.com/apache/beam/blob/269130af20462e9ffd35b74f5c4873efeea183a0/sdks/python/apache_beam/runners/worker/operations.py#L917-L930),
setup runs during operation construction. The operation tree must finish
construction before it becomes the tree that the bundle processor later shuts
down. DirectRunner also calls setup while constructing the evaluator. I have
not found a supported runner path that calls this wrapper's teardown after
failed setup.
`reset_state()` has test call sites only in the current tree. It stops the
shared event loop, so a bare `None` guard could proceed to wrapped cleanup
without waiting for active work after that reset. I am leaving the
implementation unchanged and this thread open for human review, rather than
weakening the wait-before-cleanup behavior for that test-only transition.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
