Currently we use execution-date when generating the command to send via executor in https://github.com/apache/airflow/blob/9c94b72d440b18a9e42123d20d48b951712038f9/airflow/models/taskinstance.py#L420-L436

We should replace the execution_date in their to run_id instead and handle the corresponding changes needs on the executor and worker.
tests/jobs/test_local_task_job.py
with callback_count_lock:
                failure_callback_called.value += 1
            assert context['dag_run'].dag_id == 'test_failure_callback_race'
            assert isinstance(context['exception'], AirflowFailException)
        def task_function(ti):
            raise AirflowFailException()
        with dag_maker("test_failure_callback_race"):
            task = PythonOperator(
                task_id='test_exit_on_failure',
                python_callable=task_function,
                on_failure_callback=failure_callback,
            )
        dag_maker.create_dagrun()
        ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
        ti.refresh_from_db()
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
        
        
        
        
        
        
        #
        
        
        def dummy_return_code(*args, **kwargs):