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/models/test_taskinstance.py
op1 = DummyOperator(task_id='test_op_1')
        op2 = DummyOperator(task_id='test_op_2')
        dag, op3 = create_dummy_dag(task_id='test_op_3')
        _, op4 = create_dummy_dag('dag2', task_id='test_op_4')
        
        assert [i.has_dag() for i in [op1, op2, op3, op4]] == [False, False, True, True]
        
        with pytest.raises(AirflowException):
            op1.set_downstream(op2)
        
        op1.dag = dag
        op1.set_downstream(op2)
        assert op2.dag is dag
        
        with pytest.raises(AirflowException):
            op1.set_downstream(op4)
        with pytest.raises(AirflowException):
            op1.set_downstream([op3, op4])
    def test_bitshift_compose_operators(self, dag_maker):
        with dag_maker('dag'):
            op1 = DummyOperator(task_id='test_op_1')
            op2 = DummyOperator(task_id='test_op_2')
            op3 = DummyOperator(task_id='test_op_3')
            op1 >> op2 << op3
        dag_maker.create_dagrun()
        
        assert op2 in op1.downstream_list
        assert op2 in op3.downstream_list
    @patch.object(DAG, 'get_concurrency_reached')
    def test_requeue_over_dag_concurrency(self, mock_concurrency_reached, create_dummy_dag):