Fix bug in LivyOperator when its trigger times out

[mateuslatrova]

Related issue

closes: #37898

Explain

When a LivyOperator was instantiated with deferrable=True and its batch job ran for more time than the set execution_timeout, airflow would detect this timeout and would cancel the trigger and then try to kill the task with the 'on_kill' method.

But that would fail raising an AttributeError because the batch_id attribute wouldn't be defined in the constructor method.

From now on, the LivyTrigger will timeout itself before airflow does it, and it will send an event to the LivyOperator signaling that a timeout happened. This way, the operator can stop the running Livy batch job, and can fail the task instance gracefully.

---

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[skestle]

I came across this, trying to figure out why the livy trigger had it's own timeout mechanism.
I understand that the error in the bug can be caused by un-initialised _batch_id, but I can't figure out how it would become uninitialised while using the deferred trigger.

def execute(self, context: Context) -> Any:
        self._batch_id = self.hook.post_batch(**self.spark_params)

initialised the batch id, which is passed to the trigger - why was batch id not still set when on_kill is called later (after general airflow timeout)?

[skestle]

Found on https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html:

When you opt to defer, your operator will stop executing at that point and be removed from its current worker. No state will persist, such as local variables or attributes set on self. When your operator resumes, it resumes as a new instance of it. The only way you can pass state from the old instance of the operator to the new one is with method_name and kwargs.

I just might make a PR for baseoperator.defer to highlight this functionality