add backward compatibility for an old close slice logic (#36774)

Author: lazebnyi

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/cursor.py

@@ -3,6 +3,7 @@
 #
 
 from abc import ABC, abstractmethod
+from typing import Any
 
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.declarative.types import Record, StreamSlice, StreamState
@@ -34,7 +35,7 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
         pass
 
     @abstractmethod
-    def close_slice(self, stream_slice: StreamSlice) -> None:
+    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
         """
         Update state based on the stream slice. Note that `stream_slice.cursor_slice` and `most_recent_record.associated_slice` are expected
         to be the same but we make it explicit here that `stream_slice` should be leveraged to update the state. We do not pass in the

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -138,7 +138,7 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
         ):
             self._highest_observed_cursor_field_value = record_cursor_value
 
-    def close_slice(self, stream_slice: StreamSlice) -> None:
+    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
         if stream_slice.partition:
             raise ValueError(f"Stream slice {stream_slice} should not have a partition. Got {stream_slice.partition}.")
         cursor_value_str_by_cursor_value_datetime = dict(

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/per_partition_cursor.py

@@ -91,10 +91,10 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
             StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), record
         )
 
-    def close_slice(self, stream_slice: StreamSlice) -> None:
+    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
         try:
             self._cursor_per_partition[self._to_partition_key(stream_slice.partition)].close_slice(
-                StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice)
+                StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), *args
             )
         except KeyError as exception:
             raise ValueError(

airbyte-cdk/python/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -314,6 +314,7 @@ def read_records(
         # Fixing paginator types has a long tail of dependencies
         self._paginator.reset()
 
+        most_recent_record_from_slice = None
         record_generator = partial(
             self._parse_records,
             stream_state=self.state or {},
@@ -325,10 +326,14 @@ def read_records(
             if self.cursor and current_record:
                 self.cursor.observe(_slice, current_record)
 
+            # Latest record read, not necessarily within slice boundaries.
+            # TODO Remove once all custom components implement `observe` method.
+            # https://github.com/airbytehq/airbyte-internal-issues/issues/6955
+            most_recent_record_from_slice = self._get_most_recent_record(most_recent_record_from_slice, current_record, _slice)
             yield stream_data
 
         if self.cursor:
-            self.cursor.close_slice(_slice)
+            self.cursor.close_slice(_slice, most_recent_record_from_slice)
         return
 
     def _get_most_recent_record(

airbyte-cdk/python/unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -445,7 +445,7 @@ def retriever_read_pages(_, __, ___):
         side_effect=retriever_read_pages,
     ):
         list(retriever.read_records(stream_slice=stream_slice, records_schema={}))
-        cursor.close_slice.assert_called_once_with(stream_slice)
+        cursor.close_slice.assert_called_once_with(stream_slice, first_record if first_greater_than_second else second_record)
 
 
 def test_given_stream_data_is_not_record_when_read_records_then_update_slice_with_optional_record():
@@ -478,7 +478,7 @@ def retriever_read_pages(_, __, ___):
     ):
         list(retriever.read_records(stream_slice=stream_slice, records_schema={}))
         cursor.observe.assert_not_called()
-        cursor.close_slice.assert_called_once_with(stream_slice)
+        cursor.close_slice.assert_called_once_with(stream_slice, None)
 
 
 def _generate_slices(number_of_slices):