Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. - #36367
Conversation
|
R: @scwhittle |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
scwhittle
left a comment
scwhittle
left a comment
There was a problem hiding this comment.
Is it possible to add a unit test that verifies that if processor creation throws an exception that we cleanup the channel?
|
|
||
| def cleanup(self, instruction_id): | ||
| # type: (str) -> None | ||
| for channel in self._data_channel_cache.values(): |
There was a problem hiding this comment.
do we need to worry about thread-safety with concurrent creates?
There was a problem hiding this comment.
in general, yes, it wouldn't hurt.
There was a problem hiding this comment.
although we won't likely be hitting this race since we probably create all channels relatively early
|
@tvalentyn friendly ping on comments so we can make release cut next week |
will take a look but possibly after a release cut if we want to merge this asap |
cfcc81e to
714cad6
Compare
Added one. |
714cad6 to
bf16250
Compare
2 participants
fixes: #31571
This PR resolves a deadlock in the Python SDK harness that happens when a flaky
DoFn.setup()causes an instruction failure while creating a bundle processor, which creates stuckness in the data plane, when we try to populate the input queue for the failed instruction, while the queue never cleans up, thus preventing successful element ingestion for other all other instructions.To fix, mark instruction as cleaned up in known GRPC data channels (that we cache), as we discard the instruction in
bundle_processor_cache+ add a few changes to ensure idempotentcy of the cleanup operations.Before:
After:
