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
"A -> C & B -> C, when A is QUEUED but B has FAILED, C is marked UPSTREAM_FAILED.",
            ),
        ],
    )
    def test_fast_follow(
        self, conf, dependencies, init_state, first_run_state, second_run_state, error_message, dag_maker
    ):
        with conf_vars(conf):
            session = settings.Session()
            python_callable = lambda: True
            with dag_maker('test_dagrun_fast_follow') as dag:
                task_a = PythonOperator(task_id='A', python_callable=python_callable)
                task_b = PythonOperator(task_id='B', python_callable=python_callable)
                task_c = PythonOperator(task_id='C', python_callable=python_callable)
                if 'D' in init_state:
                    task_d = PythonOperator(task_id='D', python_callable=python_callable)
                for upstream, downstream in dependencies.items():
                    dag.set_dependency(upstream, downstream)
            scheduler_job = SchedulerJob(subdir=os.devnull)
            scheduler_job.dagbag.bag_dag(dag, root_dag=dag)
            dag_run = dag.create_dagrun(run_id='test_dagrun_fast_follow', state=State.RUNNING)
            task_instance_a = TaskInstance(task_a, dag_run.execution_date, init_state['A'])
            task_instance_b = TaskInstance(task_b, dag_run.execution_date, init_state['B'])
            task_instance_c = TaskInstance(task_c, dag_run.execution_date, init_state['C'])
            if 'D' in init_state:
                task_instance_d = TaskInstance(task_d, dag_run.execution_date, init_state['D'])
                session.merge(task_instance_d)
            session.merge(task_instance_a)
            session.merge(task_instance_b)
            session.merge(task_instance_c)