Repository: apache/airflow
File: task-sdk/src/airflow/sdk/execution_time/task_runner.py (Python)

Code before:
        try:
            get_listener_manager().hook.on_task_instance_failed(
                previous_state=TaskInstanceState.RUNNING, task_instance=ti, error=error
            )
        except Exception:
            log.exception("error calling listener")
        if error and task.email_on_retry and task.email:
            _send_error_email_notification(task, ti, context, error, log)
    elif state == TaskInstanceState.FAILED:
        _run_task_state_change_callbacks(task, "on_failure_callback", context, log)
        try:
            get_listener_manager().hook.on_task_instance_failed(
                previous_state=TaskInstanceState.RUNNING, task_instance=ti, error=error
            )
        except Exception:
            log.exception("error calling listener")
        if error and task.email_on_failure and task.email:
            _send_error_email_notification(task, ti, context, error, log)

    try:
        get_listener_manager().hook.before_stopping(component=TaskRunnerMarker())
    except Exception:
        log.exception("error calling listener")


def main():
    log = structlog.get_logger(logger_name="task")

    global SUPERVISOR_COMMS
    SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor](log=log)

    stats_factory = stats_utils.get_stats_factory(Stats)
    Stats.initialize(factory=stats_factory)

    try:
        try:
            startup_details = get_startup_details()
            ti, context = startup(msg=startup_details)
        except AirflowRescheduleException as reschedule:
            log.warning("Rescheduling task during startup, marking task as UP_FOR_RESCHEDULE")
            SUPERVISOR_COMMS.send(
                msg=RescheduleTask(
                    reschedule_date=reschedule.reschedule_date,
                    end_date=datetime.now(tz=timezone.utc),
                )
            )
            sys.exit(0)
        with BundleVersionLock(
            bundle_name=ti.bundle_instance.name,
            bundle_version=ti.bundle_instance.version,
        ):

Review comment (suggestion):
main() has no body beyond the logger init. Is this a work in progress or was content accidentally removed?
(not too critical)

Code after:
        try:
            get_listener_manager().hook.on_task_instance_failed(
                previous_state=TaskInstanceState.RUNNING, task_instance=ti, error=error
            )
        except Exception:
            log.exception("error calling listener")
        if error and task.email_on_retry and task.email:
            _send_error_email_notification(task, ti, context, error, log)
    elif state == TaskInstanceState.FAILED:
        _run_task_state_change_callbacks(task, "on_failure_callback", context, log)
        try:
            get_listener_manager().hook.on_task_instance_failed(
                previous_state=TaskInstanceState.RUNNING, task_instance=ti, error=error
            )
        except Exception:
            log.exception("error calling listener")
        if error and task.email_on_failure and task.email:
            _send_error_email_notification(task, ti, context, error, log)

    try:
        get_listener_manager().hook.before_stopping(component=TaskRunnerMarker())
    except Exception:
        log.exception("error calling listener")


def main():
    log = structlog.get_logger(logger_name="task")

    global SUPERVISOR_COMMS
    SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor](log=log)

    stats_factory = stats_utils.get_stats_factory(Stats)
    Stats.initialize(factory=stats_factory)

    try:
        try:
            startup_details = get_startup_details()
            ti, context, log = startup(msg=startup_details)
        except AirflowRescheduleException as reschedule:
            log.warning("Rescheduling task during startup, marking task as UP_FOR_RESCHEDULE")
            SUPERVISOR_COMMS.send(
                msg=RescheduleTask(
                    reschedule_date=reschedule.reschedule_date,
                    end_date=datetime.now(tz=timezone.utc),
                )
            )
            sys.exit(0)
        with BundleVersionLock(
            bundle_name=ti.bundle_instance.name,
            bundle_version=ti.bundle_instance.version,
        ):