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
self.dagbag = dagbag
        with patch('airflow.jobs.base_job.sleep') as self.mock_base_job_sleep:
            yield
    def validate_ti_states(self, dag_run, ti_state_mapping, error_message):
        for task_id, expected_state in ti_state_mapping.items():
            task_instance = dag_run.get_task_instance(task_id=task_id)
            task_instance.refresh_from_db()
            assert task_instance.state == expected_state, error_message
    def test_localtaskjob_essential_attr(self, dag_maker):
        """
        Check whether essential attributes
        of LocalTaskJob can be assigned with
        proper values without intervention
        """
        with dag_maker('test_localtaskjob_essential_attr'):
            op1 = DummyOperator(task_id='op1')
        dr = dag_maker.create_dagrun()
        ti = dr.get_task_instance(task_id=op1.task_id)
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
        essential_attr = ["dag_id", "job_type", "start_date", "hostname"]
        check_result_1 = [hasattr(job1, attr) for attr in essential_attr]
        assert all(check_result_1)
        check_result_2 = [getattr(job1, attr) is not None for attr in essential_attr]
        assert all(check_result_2)
    def test_localtaskjob_heartbeat(self, dag_maker):