### Body

We have several tests marked in the code base with `@pytest.mark.quarantined`
It means that the tests are flaky and if fail in CI it does not fail the build.
The goal is to fix the tests and make them stable.

This task is to gather all of them under the same issue instead of dedicated issue per test.

- [x] [TestImpersonation](https://github.com/apache/airflow/blob/bfcae349b88fd959e32bfacd027a5be976fe2132/tests/core/test_impersonation_tests.py#L117)
- [x] [TestImpersonationWithCustomPythonPath](https://github.com/apache/airflow/blob/bfcae349b88fd959e32bfacd027a5be976fe2132/tests/core/test_impersonation_tests.py#L181)
- [x] [test_exception_propagation](https://github.com/apache/airflow/blob/76f81cd4a7433b7eeddb863b2ae6ee59176cf816/tests/jobs/test_local_task_job.py#L772)
- [x] [test_localtaskjob_maintain_heart_rate](https://github.com/apache/airflow/blob/76f81cd4a7433b7eeddb863b2ae6ee59176cf816/tests/jobs/test_local_task_job.py#L402)
- [x] [test_exception_propagation](https://github.com/apache/airflow/blob/4d0fd8ef6adc35f683c7561f05688a65fd7451f4/tests/executors/test_celery_executor.py#L103)
- [x] [test_process_sigterm_works_with_retries](https://github.com/apache/airflow/blob/65010fda091242870a410c65478eae362899763b/tests/jobs/test_local_task_job.py#L770)


### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
"""
        Test that ensures that mark_failure in the UI fails
        the task, and executes on_failure_callback
        """
        dag = get_test_dag("test_mark_state")
        with create_session() as session:
            dr = dag.create_dagrun(
                state=State.RUNNING,
                execution_date=DEFAULT_DATE,
                run_type=DagRunType.SCHEDULED,
                session=session,
            )
        task = dag.get_task(task_id="test_mark_failure_externally")
        ti = dr.get_task_instance(task.task_id)
        ti.refresh_from_task(task)
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
        with timeout(30):
            
            
            job1.run()
        ti.refresh_from_db()
        assert ti.state == State.FAILED
        assert (
            "State of this instance has been externally set to failed. Terminating instance."
        ) in caplog.text
    def test_dagrun_timeout_logged_in_task_logs(self, caplog, get_test_dag):