fix(declarative): make cursor datetime format mismatch errors actionable

airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py

@@ -11,7 +11,10 @@
 
 from airbyte_cdk.legacy.sources.declarative.incremental.declarative_cursor import DeclarativeCursor
 from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, Level, Type
-from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+    DatetimeParser,
+)
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.interpolation.jinja import JinjaInterpolation
@@ -308,12 +311,19 @@ def _get_date(
         return comparator(cursor_date, default_date)
 
     def parse_date(self, date: str) -> datetime.datetime:
-        for datetime_format in self.cursor_datetime_formats + [self.datetime_format]:
+        formats = self.cursor_datetime_formats + [self.datetime_format]
+        last_error: Optional[ValueError] = None
+        for datetime_format in formats:
             try:
                 return self._parser.parse(date, datetime_format)
-            except ValueError:
-                pass
-        raise ValueError(f"No format in {self.cursor_datetime_formats} matching {date}")
+            except ValueError as error:
+                last_error = error
+        raise DatetimeFormatMismatchError(
+            value=date,
+            formats=formats,
+            cursor_field=self.cursor_field.eval(self.config),  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
+            original_error=last_error,
+        )
 
     @classmethod
     def _parse_timedelta(cls, time_str: Optional[str]) -> Union[datetime.timedelta, Duration]:

airbyte_cdk/sources/declarative/datetime/datetime_parser.py

@@ -3,7 +3,66 @@
 #
 
 import datetime
-from typing import Union
+from typing import List, Optional, Union
+
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.utils.traced_exception import AirbyteTracedException
+
+
+class DatetimeFormatMismatchError(AirbyteTracedException, ValueError):
+    """Raised when a datetime value matches none of the configured datetime formats.
+
+    Inherits from `ValueError` so that callers which already handle datetime parsing
+    failures (multi-format fallback loops, cursor value extraction) keep working, and
+    from `AirbyteTracedException` so that the failure surfaces with a user-facing
+    message instead of a raw Python error.
+
+    The user-facing `message` is deterministic: it names the stream and cursor field
+    when they are known, and never embeds the offending value or the format tokens.
+    Those live in `internal_message`.
+    """
+
+    def __init__(
+        self,
+        value: object,
+        formats: List[str],
+        cursor_field: Optional[str] = None,
+        stream_name: Optional[str] = None,
+        original_error: Optional[BaseException] = None,
+    ) -> None:
+        self.value = value
+        self.formats = formats
+        self.cursor_field = cursor_field
+        self.stream_name = stream_name
+        super().__init__(
+            message=self._build_message(cursor_field, stream_name),
+            internal_message=f"No format in {formats} matching {value}",
+            failure_type=FailureType.system_error,
+            exception=original_error,
+        )
+
+    @staticmethod
+    def _build_message(cursor_field: Optional[str], stream_name: Optional[str]) -> str:
+        if cursor_field and stream_name:
+            return (
+                f'Cursor field "{cursor_field}" of stream "{stream_name}" matches none of the '
+                "configured datetime formats."
+            )
+        if cursor_field:
+            return f'Cursor field "{cursor_field}" matches none of the configured datetime formats.'
+        return "Datetime value matches none of the configured datetime formats."
+
+    def with_context(
+        self, cursor_field: Optional[str] = None, stream_name: Optional[str] = None
+    ) -> "DatetimeFormatMismatchError":
+        """Return an equivalent error naming the stream and cursor field."""
+        return DatetimeFormatMismatchError(
+            value=self.value,
+            formats=self.formats,
+            cursor_field=cursor_field or self.cursor_field,
+            stream_name=stream_name or self.stream_name,
+            original_error=self,
+        )
 
 
 class DatetimeParser:
@@ -35,7 +94,12 @@ def parse(self, date: Union[str, int], format: str) -> datetime.datetime:
             return self._UNIX_EPOCH + datetime.timedelta(milliseconds=int(date))
         elif "%_ms" in format:
             format = format.replace("%_ms", "%f")
-        parsed_datetime = datetime.datetime.strptime(str(date), format)
+        try:
+            parsed_datetime = datetime.datetime.strptime(str(date), format)
+        except ValueError as error:
+            raise DatetimeFormatMismatchError(
+                value=date, formats=[format], original_error=error
+            ) from error
         if self._is_naive(parsed_datetime):
             return parsed_datetime.replace(tzinfo=datetime.timezone.utc)
         return parsed_datetime

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -20,6 +20,7 @@
 )
 
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeFormatMismatchError
 from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.concurrent.clamping import ClampingStrategy, NoClamping
@@ -230,8 +231,9 @@ def __init__(
         ] = {}
         self._has_closed_at_least_one_slice = False
         self._cursor_granularity = cursor_granularity
-        # Flag to track if the logger has been triggered (per stream)
+        # Flags to track if the loggers have been triggered (per stream)
         self._should_be_synced_logger_triggered = False
+        self._unparsable_cursor_value_logger_triggered = False
         self._clamping_strategy = clamping_strategy
         self._is_ascending_order = True
 
@@ -264,6 +266,16 @@ def _slice_boundary_fields_wrapper(self) -> Tuple[str, str]:
 
     def _get_concurrent_state(
         self, state: MutableMapping[str, Any]
+    ) -> Tuple[CursorValueType, MutableMapping[str, Any]]:
+        try:
+            return self._get_concurrent_state_without_context(state)
+        except DatetimeFormatMismatchError as error:
+            raise error.with_context(
+                cursor_field=self._cursor_field.cursor_field_key, stream_name=self._stream_name
+            ) from error
+
+    def _get_concurrent_state_without_context(
+        self, state: MutableMapping[str, Any]
     ) -> Tuple[CursorValueType, MutableMapping[str, Any]]:
         if self._connector_state_converter.is_state_message_compatible(state):
             partitioned_state = self._connector_state_converter.deserialize(state)
@@ -307,6 +319,8 @@ def observe(self, record: Record) -> None:
                 self._most_recent_cursor_value_per_partition[record.associated_slice] = cursor_value
             elif most_recent_cursor_value > cursor_value:
                 self._is_ascending_order = False
+        except DatetimeFormatMismatchError:
+            self._log_for_record_with_unparsable_cursor_value()
         except ValueError:
             self._log_for_record_without_cursor_value()
 
@@ -567,6 +581,9 @@ def should_be_synced(self, record: Record) -> bool:
         """
         try:
             record_cursor_value: CursorValueType = self._extract_cursor_value(record)
+        except DatetimeFormatMismatchError:
+            self._log_for_record_with_unparsable_cursor_value()
+            return True
         except ValueError:
             self._log_for_record_without_cursor_value()
             return True
@@ -579,6 +596,13 @@ def _log_for_record_without_cursor_value(self) -> None:
             )
             self._should_be_synced_logger_triggered = True
 
+    def _log_for_record_with_unparsable_cursor_value(self) -> None:
+        if not self._unparsable_cursor_value_logger_triggered:
+            LOGGER.warning(
+                f"Cursor field `{self.cursor_field.cursor_field_key}` in record for stream {self._stream_name} matches none of the configured datetime formats. The incremental sync will assume the record needs to be synced"
+            )
+            self._unparsable_cursor_value_logger_triggered = True
+
     def reduce_slice_range(self, stream_slice: StreamSlice) -> StreamSlice:
         # In theory, we might be more flexible here meaning that it doesn't need to be in ascending order but it just
         # needs to be ordered. For now though, we will only support ascending order.

airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py

@@ -8,7 +8,10 @@
 
 # FIXME We would eventually like the Concurrent package do be agnostic of the declarative package. However, this is a breaking change and
 #  the goal in the short term is only to fix the issue we are seeing for source-declarative-manifest.
-from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+    DatetimeParser,
+)
 from airbyte_cdk.sources.streams.concurrent.cursor import CursorField
 from airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter import (
     AbstractStreamStateConverter,
@@ -215,9 +218,14 @@ def output_format(self, timestamp: datetime) -> str:
         return self._parser.format(timestamp, self._datetime_format)
 
     def parse_timestamp(self, timestamp: str) -> datetime:
+        last_error: Optional[ValueError] = None
         for datetime_format in self._input_datetime_formats:
             try:
                 return self._parser.parse(timestamp, datetime_format)
-            except ValueError:
-                pass
-        raise ValueError(f"No format in {self._input_datetime_formats} matching {timestamp}")
+            except ValueError as error:
+                last_error = error
+        raise DatetimeFormatMismatchError(
+            value=timestamp,
+            formats=self._input_datetime_formats,
+            original_error=last_error,
+        )

unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py

@@ -8,6 +8,10 @@
 import pytest
 
 from airbyte_cdk.legacy.sources.declarative.incremental import DatetimeBasedCursor
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+)
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.requesters.request_option import (
@@ -1020,9 +1024,17 @@ def test_given_unknown_format_when_parse_date_then_raise_error():
         config=config,
         parameters={},
     )
-    with pytest.raises(ValueError):
+    with pytest.raises(DatetimeFormatMismatchError) as exc_info:
         slicer.parse_date("2021-01-01T00:00:00.000000+0000")
 
+    assert isinstance(exc_info.value, ValueError)
+    assert (
+        exc_info.value.message
+        == f'Cursor field "{cursor_field}" matches none of the configured datetime formats.'
+    )
+    assert exc_info.value.failure_type == FailureType.system_error
+    assert "2021-01-01T00:00:00.000000+0000" in (exc_info.value.internal_message or "")
+
 
 @pytest.mark.parametrize(
     "test_name, input_dt, datetimeformat, datetimeformat_granularity, expected_output",

unit_tests/sources/declarative/datetime/test_datetime_parser.py

@@ -6,7 +6,11 @@
 
 import pytest
 
-from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+    DatetimeParser,
+)
 
 
 @pytest.mark.parametrize(
@@ -125,3 +129,33 @@ def test_format_datetime(input_dt: datetime.datetime, datetimeformat: str, expec
     parser = DatetimeParser()
     output_date = parser.format(input_dt, datetimeformat)
     assert output_date == expected_output
+
+
+def test_given_value_not_matching_format_when_parse_then_raise_traced_error():
+    with pytest.raises(DatetimeFormatMismatchError) as exc_info:
+        DatetimeParser().parse("2025-06-28T18:35:16Z", "%Y-%m-%dT%H:%M:%S.%fZ")
+
+    assert isinstance(exc_info.value, ValueError)
+    assert (
+        exc_info.value.message == "Datetime value matches none of the configured datetime formats."
+    )
+    assert exc_info.value.failure_type == FailureType.system_error
+    assert (
+        exc_info.value.internal_message
+        == "No format in ['%Y-%m-%dT%H:%M:%S.%fZ'] matching 2025-06-28T18:35:16Z"
+    )
+
+
+def test_given_context_when_with_context_then_message_names_stream_and_cursor_field():
+    error = DatetimeFormatMismatchError(
+        value="2025-06-28T18:35:16Z", formats=["%Y-%m-%dT%H:%M:%S.%fZ"]
+    ).with_context(cursor_field="updated_at", stream_name="items")
+
+    assert error.message == (
+        'Cursor field "updated_at" of stream "items" matches none of the configured datetime '
+        "formats."
+    )
+    assert (
+        error.internal_message
+        == "No format in ['%Y-%m-%dT%H:%M:%S.%fZ'] matching 2025-06-28T18:35:16Z"
+    )

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -24,6 +24,9 @@
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
 )
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+)
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
 
 _CONFIG = {"start_date": "2024-07-01T00:00:00.000Z"}
@@ -608,9 +611,16 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor():
         source_config=manifest, config=_CONFIG, catalog=None, state=state
     )
 
-    with pytest.raises(ValueError, match="not-a-date"):
+    with pytest.raises(DatetimeFormatMismatchError) as exc_info:
         source.discover(logger=MagicMock(), config=_CONFIG)
 
+    assert isinstance(exc_info.value, ValueError)
+    assert exc_info.value.message == (
+        'Cursor field "updated_at" of stream "TestStream" matches none of the configured '
+        "datetime formats."
+    )
+    assert "not-a-date" in (exc_info.value.internal_message or "")
+
 
 @freezegun.freeze_time("2024-07-15")
 def test_final_state_cursor_falls_back_to_full_refresh_when_state_unparseable():

unit_tests/sources/streams/concurrent/test_cursor.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import logging
 from copy import deepcopy
 from datetime import datetime, timedelta, timezone
 from typing import Any, Mapping, Optional
@@ -11,7 +12,11 @@
 import freezegun
 import pytest
 
+from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import (
+    DatetimeFormatMismatchError,
+)
 from airbyte_cdk.sources.message import MessageRepository
 from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.concurrent.clamping import (
@@ -1411,3 +1416,53 @@ def test_final_state_cursor_get_cursor_datetime_from_state_returns_now_for_no_cu
 
     result_with_empty_state = cursor.get_cursor_datetime_from_state({})
     assert result_with_empty_state is None
+
+
+_MILLIS_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+_WHOLE_SECOND_CURSOR_VALUE = "2025-06-28T18:35:16Z"
+
+
+def _cursor_with_custom_format(stream_state: Mapping[str, Any]) -> ConcurrentCursor:
+    return ConcurrentCursor(
+        _A_STREAM_NAME,
+        _A_STREAM_NAMESPACE,
+        deepcopy(stream_state),
+        Mock(spec=MessageRepository),
+        Mock(spec=ConnectorStateManager),
+        CustomFormatConcurrentStreamStateConverter(_MILLIS_FORMAT),
+        CursorField(_A_CURSOR_FIELD_KEY),
+        _SLICE_BOUNDARY_FIELDS,
+        None,
+        CustomFormatConcurrentStreamStateConverter.get_end_provider(),
+        _NO_LOOKBACK_WINDOW,
+    )
+
+
+def test_given_state_value_not_matching_format_when_create_cursor_then_raise_traced_error() -> None:
+    with pytest.raises(DatetimeFormatMismatchError) as exc_info:
+        _cursor_with_custom_format({_A_CURSOR_FIELD_KEY: _WHOLE_SECOND_CURSOR_VALUE})
+
+    assert isinstance(exc_info.value, ValueError)
+    assert exc_info.value.message == (
+        f'Cursor field "{_A_CURSOR_FIELD_KEY}" of stream "{_A_STREAM_NAME}" matches none of the '
+        "configured datetime formats."
+    )
+    assert exc_info.value.failure_type == FailureType.system_error
+    assert _WHOLE_SECOND_CURSOR_VALUE in (exc_info.value.internal_message or "")
+    assert _MILLIS_FORMAT in (exc_info.value.internal_message or "")
+
+
+def test_given_record_cursor_value_not_matching_format_when_should_be_synced_then_log_and_sync(
+    caplog,
+) -> None:
+    cursor = _cursor_with_custom_format(_NO_STATE)
+    record = Record(
+        data={_A_CURSOR_FIELD_KEY: _WHOLE_SECOND_CURSOR_VALUE},
+        associated_slice=None,
+        stream_name=_A_STREAM_NAME,
+    )
+
+    with caplog.at_level(logging.WARNING, logger="airbyte"):
+        assert cursor.should_be_synced(record) is True
+
+    assert "matches none of the configured datetime formats" in caplog.text