KevinGG commented on a change in pull request #12339:
URL: https://github.com/apache/beam/pull/12339#discussion_r464657824
##########
File path:
sdks/python/apache_beam/runners/interactive/interactive_environment.py
##########
@@ -364,26 +382,37 @@ def get_background_caching_job(self, pipeline):
"""Gets the background caching job started from the given pipeline."""
return self._background_caching_jobs.get(str(id(pipeline)), None)
+ def evict_background_caching_job(self, pipeline=None):
+ """Evicts the background caching job started from the given pipeline. Noop
+ if the given pipeline is absent from the environment. If no pipeline is
+ specified, evicts for all pipelines."""
+ if pipeline:
+ return self._background_caching_jobs.pop(str(id(pipeline)), None)
Review comment:
That should not be necessary.
The idea is to let `cleanup` clear all states and `evict_*` functions only
do eviction for each field.
Because if we `stop` a test stream, probably, we'll also have to stop the
background caching job and any other in-memory states that is related. There is
rarely a case that one of the fields get evicted without clearing other fields
except in some unit tests, for example:
```
# In tests:
evict_xxx()
# Then in teardown, cleanup.
```
This also makes sure cleanup code doesn't redundantly show up in multiple
places or cause infinite loop when evicting/cleanup are called by different
modules.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
