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
self.hostname = get_hostname()
        self.pid = os.getpid()
        session.merge(self)
        session.commit()
        actual_start_date = timezone.utcnow()
        Stats.incr(f'ti.start.{task.dag_id}.{task.task_id}')
        try:
            if not mark_success:
                context = self.get_template_context()
                self._prepare_and_execute_task_with_callbacks(context, task)
            self.refresh_from_db(lock_for_update=True)
            self.state = State.SUCCESS
        except TaskDeferred as defer:
            
            
            self._defer_task(defer=defer)
            self.log.info(
                'Pausing task as DEFERRED. dag_id=%s, task_id=%s, execution_date=%s, start_date=%s',
                self.dag_id,
                self.task_id,
                self._date_or_empty('execution_date'),
                self._date_or_empty('start_date'),
            )
            if not test_mode:
                session.add(Log(self.state, self))
                session.merge(self)
            session.commit()
            return
        except AirflowSmartSensorException as e:
            self.log.info(e)