Recent changes to timezone support broke some operators

[ferruzzi]

Apache Airflow version

2.7.1

What happened

The Dynamo-to-S3 system test has been failing since #34492 got merged. It looks like the get_export_time task is returning a datetime in a format that is no longer being supported. This might be a matter of just converting the format of that return value, but it ma also indicate a larger issue with that change.

Relevant logs for the failing test runs:

INFO  [airflow.task] Executing <Task(DynamoDBToS3Operator): backup_db_to_point_in_time> on 2021-01-01 00:00:00+00:00
--
ERROR [airflow.task.operators] Exception rendering Jinja template for task 'backup_db_to_point_in_time', field 'export_time'. Template: XComArg(<Task(_PythonDecoratedOperator): get_export_time>)
Traceback (most recent call last):
File "/opt/airflow/airflow/models/abstractoperator.py", line 699, in _do_render_template_fields
rendered_content = self.render_template(
File "/opt/airflow/airflow/template/templater.py", line 157, in render_template
return value.resolve(context)
File "/opt/airflow/airflow/utils/session.py", line 79, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/xcom_arg.py", line 417, in resolve
result = ti.xcom_pull(
File "/opt/airflow/airflow/utils/session.py", line 76, in wrapper
return func(*args, **kwargs)
File "/opt/airflow/airflow/models/taskinstance.py", line 2967, in xcom_pull
return XCom.deserialize_value(first)
File "/opt/airflow/airflow/models/xcom.py", line 696, in deserialize_value
return BaseXCom._deserialize_value(result, False)
File "/opt/airflow/airflow/models/xcom.py", line 689, in _deserialize_value
return json.loads(result.value.decode("UTF-8"), cls=XComDecoder, object_hook=object_hook)
File "/usr/local/lib/python3.8/json/__init__.py", line 370, in loads
return cls(**kw).decode(s)
File "/usr/local/lib/python3.8/json/decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
File "/usr/local/lib/python3.8/json/decoder.py", line 353, in raw_decode
obj, end = self.scan_once(s, idx)
File "/opt/airflow/airflow/utils/json.py", line 117, in object_hook
return deserialize(dct)
File "/opt/airflow/airflow/serialization/serde.py", line 255, in deserialize
return _deserializers[classname].deserialize(classname, version, deserialize(value))
File "/opt/airflow/airflow/serialization/serializers/datetime.py", line 86, in deserialize
tz = deserialize_timezone(data[TIMEZONE][1], data[TIMEZONE][2], data[TIMEZONE][0])
File "/opt/airflow/airflow/serialization/serializers/timezone.py", line 72, in deserialize
return timezone(data)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/__init__.py", line 37, in timezone
tz = _Timezone(name, extended=extended)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/timezone.py", line 40, in __init__
tz = read(name, extend=extended)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/zoneinfo/__init__.py", line 9, in read
return Reader(extend=extend).read_for(name)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/zoneinfo/reader.py", line 50, in read_for
file_path = pytzdata.tz_path(timezone)
File "/usr/local/lib/python3.8/site-packages/pytzdata/__init__.py", line 63, in tz_path
raise ValueError('Invalid timezone')
ValueError: Invalid timezone
ERROR [airflow.models.taskinstance.TaskInstance] Task failed with exception
Traceback (most recent call last):
File "/opt/airflow/airflow/models/taskinstance.py", line 2246, in _run_raw_task
self._execute_task_with_callbacks(context, test_mode, session=session)
File "/opt/airflow/airflow/models/taskinstance.py", line 2375, in _execute_task_with_callbacks
task_orig = self.render_templates(context=context)
File "/opt/airflow/airflow/models/taskinstance.py", line 2787, in render_templates
original_task.render_template_fields(context)
File "/opt/airflow/airflow/models/baseoperator.py", line 1248, in render_template_fields
self._do_render_template_fields(self, self.template_fields, context, jinja_env, set())
File "/opt/airflow/airflow/utils/session.py", line 79, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/abstractoperator.py", line 699, in _do_render_template_fields
rendered_content = self.render_template(
File "/opt/airflow/airflow/template/templater.py", line 157, in render_template
return value.resolve(context)
File "/opt/airflow/airflow/utils/session.py", line 79, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/xcom_arg.py", line 417, in resolve
result = ti.xcom_pull(
File "/opt/airflow/airflow/utils/session.py", line 76, in wrapper
return func(*args, **kwargs)
File "/opt/airflow/airflow/models/taskinstance.py", line 2967, in xcom_pull
return XCom.deserialize_value(first)
File "/opt/airflow/airflow/models/xcom.py", line 696, in deserialize_value
return BaseXCom._deserialize_value(result, False)
File "/opt/airflow/airflow/models/xcom.py", line 689, in _deserialize_value
return json.loads(result.value.decode("UTF-8"), cls=XComDecoder, object_hook=object_hook)
File "/usr/local/lib/python3.8/json/__init__.py", line 370, in loads
return cls(**kw).decode(s)
File "/usr/local/lib/python3.8/json/decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
File "/usr/local/lib/python3.8/json/decoder.py", line 353, in raw_decode
obj, end = self.scan_once(s, idx)
File "/opt/airflow/airflow/utils/json.py", line 117, in object_hook
return deserialize(dct)
File "/opt/airflow/airflow/serialization/serde.py", line 255, in deserialize
return _deserializers[classname].deserialize(classname, version, deserialize(value))
File "/opt/airflow/airflow/serialization/serializers/datetime.py", line 86, in deserialize
tz = deserialize_timezone(data[TIMEZONE][1], data[TIMEZONE][2], data[TIMEZONE][0])
File "/opt/airflow/airflow/serialization/serializers/timezone.py", line 72, in deserialize
return timezone(data)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/__init__.py", line 37, in timezone
tz = _Timezone(name, extended=extended)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/timezone.py", line 40, in __init__
tz = read(name, extend=extended)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/zoneinfo/__init__.py", line 9, in read
return Reader(extend=extend).read_for(name)
File "/usr/local/lib/python3.8/site-packages/pendulum/tz/zoneinfo/reader.py", line 50, in read_for
file_path = pytzdata.tz_path(timezone)
File "/usr/local/lib/python3.8/site-packages/pytzdata/__init__.py", line 63, in tz_path
raise ValueError('Invalid timezone')
ValueError: Invalid timezone

cc @hussein-awala

What you think should happen instead

The PR I referenced broke existing code. If this is a matter of adjusting one return value that's great, but it may indicate a larger issue.

How to reproduce

The issue comes up when running the system test, there are surely other ways to repro it.

Operating System

linux

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

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

[bolkedebruin]

Can we be more clear on what exactly was the 'invalid timezone'? The tests were passing at the time of merging. So a debug statement would be great.

It looks weird, because the deserialized value should just be a normal value.

[Taragolis]

This about Amazon system tests, we do not run any system tests into the Airflow CI this tests intend to run by owners of the providers or as examples for the documentation. AFAIK some of the Amazon system test run into Amazon CI on actual AWS environments and provide some nice Dashboard

As I can see, this task return the datetime, however timezone which returned by boto3 use dateutil tzinfo implementation, which can not be serialized

airflow/tests/system/providers/amazon/aws/example_dynamodb_to_s3.py

Lines 85 to 93 in 84e6036

	@task
	def get_export_time(table_name: str):
	r = boto3.client("dynamodb").describe_continuous_backups(
	TableName=table_name,
	)
	return r["ContinuousBackupsDescription"]["PointInTimeRecoveryDescription"]["EarliestRestorableDateTime"]

and the deserialise error happen int this task

airflow/tests/system/providers/amazon/aws/example_dynamodb_to_s3.py

Lines 165 to 174 in 84e6036

	# [START howto_transfer_dynamodb_to_s3_in_some_point_in_time]
	backup_db_to_point_in_time = DynamoDBToS3Operator(
	task_id="backup_db_to_point_in_time",
	dynamodb_table_name=table_name,
	file_size=1000,
	s3_bucket_name=bucket_name,
	export_time=export_time,
	s3_key_prefix=f"{S3_KEY_PREFIX}-3-",
	)
	# [END howto_transfer_dynamodb_to_s3_in_some_point_in_time]

It works before, because previous (incorrect) implementation return UTC when we call tzname() during serialisation.

---

Simple snippet for show what is going wrong

from dateutil.tz import tzutc
from datetime import datetime

from airflow.serialization.serializers.datetime import serialize, deserialize


sample_data = datetime(2021, 1, 1, tzinfo=tzutc())

ser_data = serialize(sample_data)
de_data = deserialize(classname=ser_data[1], version=ser_data[2], data=ser_data[0])  # Error happen here

---

This might be fixed in system tests if convert it to Pendulum Timezone by use convert_to_utc, at least it works locally. cc: @ferruzzi

from datetime import datetime
from airflow.utils.timezone import convert_to_utc
from airflow.serialization.serializers.datetime import serialize, deserialize
from zoneinfo import ZoneInfo

sample_data = datetime(2021, 1, 1, tzinfo=ZoneInfo("UTC"))
sample_data = convert_to_utc(sample_data)  # <- This will convert tz to pendulum timezone

ser_data = serialize(sample_data)
de_data = deserialize(classname=ser_data[1], version=ser_data[2], data=ser_data[0])

[bolkedebruin]

I dont think that is the right approach as that would convert the timezone to UTC even if it is not:

sample_data = datetime(2021, 1, 1, tzinfo=ZoneInfo("Europe/Paris"))
sample_data = convert_to_utc(sample_data)  # <- This will convert tz to pendulum timezone

ser_data = serialize(sample_data)

print(ser_data)

({'timestamp': 1609455600.0, 'tz': ('UTC', 'pendulum.tz.timezone.FixedTimezone', 1, True)}, 'pendulum.datetime.DateTime', 2, True)

I think the best way forward is to just extend the timezone serializer to deal with ZoneInfo and pytz. Relying on pendulum.instance() could also work, but that would convert the timezone to a FixedTimezone.

This is from Pendulum master:

 def _get_tzinfo_name(tzinfo: datetime.tzinfo | None) -> str | None:
    if tzinfo is None:
        return None

    if hasattr(tzinfo, "key"):
        # zoneinfo timezone
        return cast(zoneinfo.ZoneInfo, tzinfo).key
    elif hasattr(tzinfo, "name"):
        # Pendulum timezone
        return cast(Timezone, tzinfo).name
    elif hasattr(tzinfo, "zone"):
        # pytz timezone
        return tzinfo.zone  # type: ignore[no-any-return]

    return None

[Taragolis]

I dont think that is the right approach as that would convert the timezone to UTC even if it is not:

That was suggest to Amazon system tests because in this case it would be always UTC.

upd: Just for clarify why it would be always UTC in this case
https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_DescribeContinuousBackups.html return epoch and botocore always assume that epoch in UTC and add dateutil.tz.tzutc as tzinfo

[ferruzzi]

That was suggest to Amazon system tests because in this case it would be always UTC.

Sure, but the more important question is what else was broken here? We can adjust the one test easy enough, but is that really the right answer?

[Taragolis]

Sure, but the more important question is what else was broken here?

I think there are plenty things might be broken but we do not know right now. And it is not a joke, it is regular things when it comes to datetimes and things doesn't become easier when we also need to handle serde in top of that 😢

We can adjust the one test easy enough, but is that really the right answer?

It was a suggestion for a quick fix but I hope @bolkedebruin 's PR also fix this test

[bolkedebruin]

This should be fixed in main now.