TimeSensorAsync breaks if target_time is timezone-aware

[Gollum999]

Apache Airflow version

2.3.2 (latest released)

What happened

TimeSensorAsync fails with the following error if target_time is aware:

[2022-06-29, 05:09:11 CDT] {taskinstance.py:1889} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/sensors/time_sensor.py", line 60, in execute
    trigger=DateTimeTrigger(moment=self.target_datetime),
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/triggers/temporal.py", line 42, in __init__
    raise ValueError(f"The passed datetime must be using Pendulum's UTC, not {moment.tzinfo!r}")
ValueError: The passed datetime must be using Pendulum's UTC, not Timezone('America/Chicago')

What you think should happen instead

Given the fact that TimeSensor correctly handles timezones (#9882), this seems like a bug. TimeSensorAsync should be a drop-in replacement for TimeSensor, and therefore should have the same timezone behavior.

How to reproduce

#!/usr/bin/env python3
import datetime

from airflow.decorators import dag
from airflow.sensors.time_sensor import TimeSensor, TimeSensorAsync
import pendulum


@dag(
    start_date=datetime.datetime(2022, 6, 29),
    schedule_interval='@daily',
)
def time_sensor_dag():
    naive_time1 = datetime.time( 0,  1)
    aware_time1 = datetime.time( 0,  1).replace(tzinfo=pendulum.local_timezone())
    naive_time2 = pendulum.time(23, 59)
    aware_time2 = pendulum.time(23, 59).replace(tzinfo=pendulum.local_timezone())

    TimeSensor(task_id='naive_time1', target_time=naive_time1, mode='reschedule')
    TimeSensor(task_id='naive_time2', target_time=naive_time2, mode='reschedule')
    TimeSensor(task_id='aware_time1', target_time=aware_time1, mode='reschedule')
    TimeSensor(task_id='aware_time2', target_time=aware_time2, mode='reschedule')

    TimeSensorAsync(task_id='async_naive_time1', target_time=naive_time1)
    TimeSensorAsync(task_id='async_naive_time2', target_time=naive_time2)
    TimeSensorAsync(task_id='async_aware_time1', target_time=aware_time1)  # fails
    TimeSensorAsync(task_id='async_aware_time2', target_time=aware_time2)  # fails

dag = time_sensor_dag()

This can also happen if the target_time is naive and core.default_timezone = system.

Operating System

CentOS Stream 8

Versions of Apache Airflow Providers

N/A

Deployment

Other

Deployment details

Standalone

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[uranusjr]

Perhaps we could allow all Pendulum timezones instead of requiring UTC. Not sure.

[uranusjr]

Convert to a feature since not allowing non-UTC is an explicit design choice made in the current implementation. Need some context on this to decide what best to do here @andrewgodwin

[andrewgodwin]

I don't see a problem with adding timezone conversion code to convert it to UTC at call time - I just add UTC-only checks to all timezone items by default if I didn't specifically write timezone support, so there's a nice clean error rather than weird behaviour.