From dc229deeb0ab9ed2ef34a21f69e73fb482c9286a Mon Sep 17 00:00:00 2001 From: Mariano Benedettini Date: Fri, 24 Jul 2026 14:40:46 -0300 Subject: [PATCH] fix: normalize timezone-aware cursor values to UTC before serializing state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DatetimeParser.format() and the concurrent state converters serialized a timezone-aware cursor datetime using its local wall-clock components, appending a literal `Z` without converting to UTC. A commit timestamp like 2026-07-23T21:59:38+05:30 was stored as 2026-07-23T21:59:38Z instead of the true 2026-07-23T16:29:38Z. For a stream using a global client-side incremental cursor (global_substream_cursor + is_client_side_incremental), this relabeled value wins the global-cursor max and is then used, minus the runtime lookback, as the client-side filter floor — silently dropping legitimate records whose true instant is earlier than the corrupted floor. Observed in source-gitlab's merge_request_commits stream. Fix: normalize datetime values to UTC before serialization in the state converters' output_format (CustomFormat and IsoMillis), treating naive datetimes as UTC to match the existing parse()/from_datetime() convention. This also makes the global-cursor string comparison chronologically correct for fixed-width UTC formats. Related: #1014. --- .../datetime_stream_state_converter.py | 14 ++++- .../test_datetime_state_converter.py | 51 ++++++++++++++++++- 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py index fdb5d4d774..d0cf126b9a 100644 --- a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py +++ b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py @@ -42,6 +42,16 @@ def increment(self, timestamp: datetime) -> datetime: ... @abstractmethod def parse_timestamp(self, timestamp: Any) -> datetime: ... + @staticmethod + def _to_utc(timestamp: datetime) -> datetime: + # Serialized state must represent an absolute instant in UTC. A timezone-aware + # datetime carrying a non-UTC offset would otherwise be formatted with its local + # wall-clock components, relabeling e.g. 21:59:38+05:30 as 21:59:38Z. Naive values + # are treated as UTC to match parse()/from_datetime() conventions. + if timestamp.tzinfo is None: + return timestamp.replace(tzinfo=timezone.utc) + return timestamp.astimezone(timezone.utc) + @abstractmethod def output_format(self, timestamp: datetime) -> Any: ... @@ -176,7 +186,7 @@ def output_format(self, timestamp: datetime) -> str: Returns: str: ISO8601/RFC3339 formatted string with milliseconds. """ - dt = AirbyteDateTime.from_datetime(timestamp) + dt = AirbyteDateTime.from_datetime(self._to_utc(timestamp)) # Always include milliseconds, even if zero millis = dt.microsecond // 1000 if dt.microsecond else 0 return f"{dt.year:04d}-{dt.month:02d}-{dt.day:02d}T{dt.hour:02d}:{dt.minute:02d}:{dt.second:02d}.{millis:03d}Z" @@ -212,7 +222,7 @@ def __init__( self._parser = DatetimeParser() def output_format(self, timestamp: datetime) -> str: - return self._parser.format(timestamp, self._datetime_format) + return self._parser.format(self._to_utc(timestamp), self._datetime_format) def parse_timestamp(self, timestamp: str) -> datetime: for datetime_format in self._input_datetime_formats: diff --git a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py index ebcb3c20c4..4d17aebe6b 100644 --- a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py +++ b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py @@ -2,7 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import pytest @@ -406,3 +406,52 @@ def test_given_when_parse_timestamp_then_eventually_fallback_on_output_format(): parsed_datetime = converter.parse_timestamp("2024-01-01T02:00:00") assert parsed_datetime == datetime(2024, 1, 1, 2, 0, 0, tzinfo=timezone.utc) + + +@pytest.mark.parametrize( + "converter,timestamp,expected_output", + [ + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"), + datetime(2026, 7, 23, 21, 59, 38, tzinfo=timezone(timedelta(hours=5, minutes=30))), + "2026-07-23T16:29:38Z", + id="custom-format-positive-offset-normalized-to-utc", + ), + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"), + datetime(2026, 7, 23, 12, 1, 25, tzinfo=timezone(timedelta(hours=-5))), + "2026-07-23T17:01:25Z", + id="custom-format-negative-offset-normalized-to-utc", + ), + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"), + datetime(2026, 7, 23, 16, 29, 38, tzinfo=timezone.utc), + "2026-07-23T16:29:38Z", + id="custom-format-utc-unchanged", + ), + pytest.param( + CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"), + datetime(2026, 7, 23, 16, 29, 38), + "2026-07-23T16:29:38Z", + id="custom-format-naive-treated-as-utc", + ), + pytest.param( + IsoMillisConcurrentStreamStateConverter(), + datetime(2026, 7, 23, 21, 59, 38, tzinfo=timezone(timedelta(hours=5, minutes=30))), + "2026-07-23T16:29:38.000Z", + id="iso-millis-positive-offset-normalized-to-utc", + ), + pytest.param( + IsoMillisConcurrentStreamStateConverter(), + datetime(2026, 7, 23, 16, 29, 38), + "2026-07-23T16:29:38.000Z", + id="iso-millis-naive-treated-as-utc", + ), + ], +) +def test_output_format_normalizes_offset_aware_datetime_to_utc( + converter, timestamp, expected_output +): + """A cursor value carrying a non-UTC offset must be serialized as its absolute UTC + instant, not its local wall clock relabeled with `Z`.""" + assert converter.output_format(timestamp) == expected_output