[Taragolis]

Unfortunately this test still not stable and first run 🤦‍♂️ 🤦‍♂️ 🤦‍♂️ in main failed again.

https://github.com/apache/airflow/actions/runs/3971462849/jobs/6808560333:

        finally:
            proc.kill()
    
        assert retry_callback_called.value == 1
        with create_session() as session:
            ti = (
                session.query(TaskInstance)
                .filter(
                    TaskInstance.dag_id == dag_id,
                    TaskInstance.task_id == task_id,
                    TaskInstance.run_id == run_id,
                )
                .one()
            )
>       assert ti.state == State.UP_FOR_RETRY
E       AssertionError: assert 'running' == <TaskInstance...up_for_retry'>
E         - up_for_retry
E         + running

tests/jobs/test_local_task_job.py:895: AssertionError

This error basically mean that we send SIGKILL proc.kill() before state changed but callback already executed.

Note: We do not check state in the previous version of the test, only callback result

[Taragolis]

Oh... status check not make any sense, we initially call all callbacks and only after that we commit to DB.

airflow/airflow/models/taskinstance.py

Lines 1859 to 1870 in 49ad77c

	if task and email_for_state(task) and task.email:
	try:
	self.email_alert(error, task)
	except Exception:
	self.log.exception("Failed to send email to: %s", task.email)
	if callbacks and context:
	self._run_finished_callback(callbacks, context, callback_type)
	if not test_mode:
	session.merge(self)
	session.flush()