You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "tvalentyn (via GitHub)" <gi...@apache.org> on 2025/10/03 01:58:40 UTC
[PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
tvalentyn opened a new pull request, #36367:
URL: https://github.com/apache/beam/pull/36367
Fixes: #31571
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36367:
URL: https://github.com/apache/beam/pull/36367#issuecomment-3404216124
> Is it possible to add a unit test that verifies that if processor creation throws an exception that we cleanup the channel?
will take a look but possibly after a release cut if we want to merge this asap
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36367:
URL: https://github.com/apache/beam/pull/36367#discussion_r2430903996
##########
sdks/python/apache_beam/runners/worker/data_plane.py:
##########
@@ -854,6 +860,11 @@ def close(self):
channel.close()
self._data_channel_cache.clear()
+ def cleanup(self, instruction_id):
+ # type: (str) -> None
+ for channel in self._data_channel_cache.values():
Review Comment:
in general, yes, it wouldn't hurt.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36367:
URL: https://github.com/apache/beam/pull/36367#issuecomment-3363897505
R: @scwhittle
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36367:
URL: https://github.com/apache/beam/pull/36367#discussion_r2405273187
##########
sdks/python/apache_beam/runners/worker/data_plane.py:
##########
@@ -502,7 +502,7 @@ def _clean_receiving_queue(self, instruction_id):
instruction_id cannot be reused for new queue.
"""
with self._receive_lock:
- self._received.pop(instruction_id)
+ self._received.pop(instruction_id, None)
Review Comment:
add a comment on why it might not be present?
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -559,15 +559,18 @@ def discard(self, instruction_id, exception):
"""
Marks the instruction id as failed shutting down the ``BundleProcessor``.
"""
+ processor = None
with self._lock:
self.failed_instruction_ids[instruction_id] = exception
while len(self.failed_instruction_ids) > MAX_FAILED_INSTRUCTIONS:
self.failed_instruction_ids.popitem(last=False)
- processor = self.active_bundle_processors[instruction_id][1]
- del self.active_bundle_processors[instruction_id]
+ if instruction_id in self.active_bundle_processors:
+ processor = self.active_bundle_processors.pop(instruction_id)[1]
Review Comment:
To be more future proof here it seems like you should change the local processor variable to the full list, and then iterate over it to shutdown below.
Alternatively, is there a reason the value in the dict is a list instead of single processor and could that be changed?
##########
sdks/python/apache_beam/runners/worker/data_plane.py:
##########
@@ -854,6 +860,11 @@ def close(self):
channel.close()
self._data_channel_cache.clear()
+ def cleanup(self, instruction_id):
+ # type: (str) -> None
+ for channel in self._data_channel_cache.values():
Review Comment:
do we need to worry about thread-safety with concurrent creates?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #36367:
URL: https://github.com/apache/beam/pull/36367#issuecomment-3388807286
@tvalentyn friendly ping on comments so we can make release cut next week
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36367:
URL: https://github.com/apache/beam/pull/36367#issuecomment-3363899051
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36367:
URL: https://github.com/apache/beam/pull/36367#discussion_r2430902730
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -559,15 +559,18 @@ def discard(self, instruction_id, exception):
"""
Marks the instruction id as failed shutting down the ``BundleProcessor``.
"""
+ processor = None
with self._lock:
self.failed_instruction_ids[instruction_id] = exception
while len(self.failed_instruction_ids) > MAX_FAILED_INSTRUCTIONS:
self.failed_instruction_ids.popitem(last=False)
- processor = self.active_bundle_processors[instruction_id][1]
- del self.active_bundle_processors[instruction_id]
+ if instruction_id in self.active_bundle_processors:
+ processor = self.active_bundle_processors.pop(instruction_id)[1]
Review Comment:
it is actually a tuple, not a list.
```
self.active_bundle_processors = { } # type: Dict[str, Tuple[str, bundle_processor.BundleProcessor]]
# sample usage
self.active_bundle_processors[ instruction_id] = bundle_descriptor_id, processor
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn merged PR #36367:
URL: https://github.com/apache/beam/pull/36367
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36367:
URL: https://github.com/apache/beam/pull/36367#issuecomment-3408338420
> Is it possible to add a unit test that verifies that if processor creation throws an exception that we cleanup the channel?
Added one.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Always mark the instruction as cleaned up in the GRPC data channel when processing an instruction fails. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36367:
URL: https://github.com/apache/beam/pull/36367#discussion_r2430904987
##########
sdks/python/apache_beam/runners/worker/data_plane.py:
##########
@@ -854,6 +860,11 @@ def close(self):
channel.close()
self._data_channel_cache.clear()
+ def cleanup(self, instruction_id):
+ # type: (str) -> None
+ for channel in self._data_channel_cache.values():
Review Comment:
although we won't likely be hitting this race since we probably create all channels relatively early
##########
sdks/python/apache_beam/runners/worker/data_plane.py:
##########
@@ -502,7 +502,7 @@ def _clean_receiving_queue(self, instruction_id):
instruction_id cannot be reused for new queue.
"""
with self._receive_lock:
- self._received.pop(instruction_id)
+ self._received.pop(instruction_id, None)
Review Comment:
done, thanks
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org