emit updated parent before last record, not after

Author: maxime.c

airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

@@ -249,6 +249,8 @@ def stream_slices(self) -> Iterable[StreamSlice]:
 
                         if is_last_record_in_slice:
                             parent_stream.cursor.close_partition(partition)
+                            if is_last_slice:
+                                parent_stream.cursor.ensure_at_least_one_state_emitted()
 
                         yield StreamSlice(
                             partition={

unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py

@@ -1119,6 +1119,8 @@ def run_incremental_parent_state_test(
 
         # Assert that the number of intermediate states is as expected
         assert len(intermediate_states) - 1 == num_intermediate_states
+        # Assert that ensure_at_least_one_state_emitted is called before yielding the last record from the last slice
+        assert intermediate_states[-1][0].stream.stream_state.__dict__["parent_state"] == intermediate_states[-2][0].stream.stream_state.__dict__["parent_state"]
 
         # For each intermediate state, perform another read starting from that state
         for state, records_before_state in intermediate_states[:-1]: