add logging to concurrent cursor

Author: maxime.c

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -4,6 +4,7 @@
 
 import functools
 import logging
+import os
 from abc import ABC, abstractmethod
 from typing import (
     Any,
@@ -237,12 +238,18 @@ def _extract_cursor_value(self, record: Record) -> Any:
         return self._connector_state_converter.parse_value(self._cursor_field.extract_value(record))
 
     def close_partition(self, partition: Partition) -> None:
+        test_env = os.getenv("PYTEST_CURRENT_TEST")
+        if test_env and "test_concurrent_declarative_source.py" in test_env:
+            LOGGER.info(f"Closing partition {partition.to_slice()}")
+            LOGGER.info(f"\tstate before is {self._concurrent_state}")
         slice_count_before = len(self._concurrent_state.get("slices", []))
         self._add_slice_to_state(partition)
         if slice_count_before < len(
             self._concurrent_state["slices"]
         ):  # only emit if at least one slice has been processed
             self._merge_partitions()
+            if test_env and "test_concurrent_declarative_source.py" in test_env:
+                LOGGER.info(f"\tstate after merged partition is {self._concurrent_state}")
             self._emit_state_message()
         self._has_closed_at_least_one_slice = True