self.job_id = job_id
        self.hostname = get_hostname()
        self.pid = os.getpid()
        if not test_mode:
            session.merge(self)
            session.commit()
        actual_start_date = timezone.utcnow()
        Stats.incr(f"ti.start.{self.task.dag_id}.{self.task.task_id}", tags=self.stats_tags)
        
        Stats.incr("ti.start", tags=self.stats_tags)
        
        for state in State.task_states:
            Stats.incr(
                f"ti.finish.{self.task.dag_id}.{self.task.task_id}.{state}",
                count=0,
                tags=self.stats_tags,
            )
            
            Stats.incr(
                "ti.finish",
                count=0,
                tags={**self.stats_tags, "state": str(state)},
            )
        with set_current_task_instance_session(session=session):
            self.task = self.task.prepare_for_execution()
            context = self.get_template_context(ignore_param_exceptions=False)
            try:
                if not mark_success:
                    self._execute_task_with_callbacks(context, test_mode, session=session)
                if not test_mode: