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
finally:
            Stats.incr(f'ti.finish.{task.dag_id}.{task.task_id}.{self.state}')
        
        self.end_date = timezone.utcnow()
        self._log_state()
        self.set_duration()
        if not test_mode:
            session.add(Log(self.state, self))
            session.merge(self)
        session.commit()
    def _prepare_and_execute_task_with_callbacks(self, context, task):
        """Prepare Task for Execution"""
        from airflow.models.renderedtifields import RenderedTaskInstanceFields
        task_copy = task.prepare_for_execution()
        self.task = task_copy
        def signal_handler(signum, frame):
            self.log.error("Received SIGTERM. Terminating subprocesses.")
            task_copy.on_kill()
            raise AirflowException("Task received SIGTERM signal")
        signal.signal(signal.SIGTERM, signal_handler)
        
        self.clear_xcom_data()
        with Stats.timer(f'dag.{task_copy.dag_id}.{task_copy.task_id}.duration'):
            self.render_templates(context=context)
            RenderedTaskInstanceFields.write(RenderedTaskInstanceFields(ti=self, render_templates=False))
            RenderedTaskInstanceFields.delete_old_records(self.task_id, self.dag_id)
            
            airflow_context_vars = context_to_airflow_vars(context, in_env_var_format=True)
            self.log.info(
                "Exporting the following env vars:\n%s",