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
dag, task1 = create_dummy_dag(
            dag_id='test_xcom',
            task_id='test_xcom_1',
            schedule_interval='@monthly',
            start_date=timezone.datetime(2016, 6, 1, 0, 0, 0),
        )
        exec_date = DEFAULT_DATE
        
        ti1 = TI(task=task1, execution_date=exec_date)
        ti1.xcom_push(key='foo', value='bar')
        
        task2 = DummyOperator(task_id='test_xcom_2', dag=dag)
        ti2 = TI(task=task2, execution_date=exec_date)
        ti2.xcom_push(key='foo', value='baz')
        
        result = ti1.xcom_pull()
        assert result is None
        
        result = ti1.xcom_pull(key='foo')
        assert result in 'baz'
        
        result = ti1.xcom_pull(task_ids='test_xcom_1', key='foo')
        assert result == 'bar'
        
        result = ti1.xcom_pull(task_ids='test_xcom_2', key='foo')
        assert result == 'baz'
        
        result = ti1.xcom_pull(task_ids=['test_xcom_1', 'test_xcom_2'], key='foo')
        assert result == ['bar', 'baz']
    def test_xcom_pull_after_success(self, create_dummy_dag):