Apply warm shutdown across coordinators #69188

Body

Python coordinator implements this in #69034. We should make this also available for other coordinators.

A small-scale refactoring to move the context manager to ActivityProcess so we can do something like:

class ActivitySubprocess:
    @classmethod
    def run(cls, **kwargs):
        with _warm_shutdown_signals():
            self = cls.start(**kwargs)
            exit_code = self.wait()
        return exit_code, self

and just use this instead of calling things separately. All coordinators can switch to use this new function. Language side shoould be fine as-is; dedicated signal handling improvements can be made but they are not specific to this.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.