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.
airflow/models/taskinstance.py
:type dag_id: str
        :param include_prior_dates: If False, only XComs from the current
            execution_date are returned. If True, XComs from previous dates
            are returned as well.
        :type include_prior_dates: bool
        :param session: Sqlalchemy ORM Session
        :type session: Session
        """
        if dag_id is None:
            dag_id = self.dag_id
        query = XCom.get_many(
            execution_date=self.execution_date,
            key=key,
            dag_ids=dag_id,
            task_ids=task_ids,
            include_prior_dates=include_prior_dates,
            session=session,
        )
        
        
        
        if is_container(task_ids):
            vals_kv = {
                result.task_id: XCom.deserialize_value(result)
                for result in query.with_entities(XCom.task_id, XCom.value)
            }
            values_ordered_by_id = [vals_kv.get(task_id) for task_id in task_ids]
            return values_ordered_by_id
        else:
            xcom = query.with_entities(XCom.value).first()