"""Prepare Task for Execution."""
        from airflow.models.renderedtifields import RenderedTaskInstanceFields
        parent_pid = os.getpid()
        def signal_handler(signum, frame):
            pid = os.getpid()
            
            
            
            
            if pid != parent_pid:
                os._exit(1)
                return
            self.log.error("Received SIGTERM. Terminating subprocesses.")
            self.task.on_kill()
            raise AirflowException("Task received SIGTERM signal")
        signal.signal(signal.SIGTERM, signal_handler)
        
        if not self.next_method:
            self.clear_xcom_data()
        with Stats.timer(f"dag.{self.task.dag_id}.{self.task.task_id}.duration", tags=self.stats_tags):
            
            self.task.params = context["params"]
            with set_current_context(context):
                task_orig = self.render_templates(context=context)
            if not test_mode:
                rtif = RenderedTaskInstanceFields(ti=self, render_templates=False)
                RenderedTaskInstanceFields.write(rtif)
                RenderedTaskInstanceFields.delete_old_records(self.task_id, self.dag_id)