log every item in the queue

Author: maxime.c

airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py

@@ -155,9 +155,6 @@ def on_record(self, record: Record) -> Iterable[AirbyteMessage]:
                 )
             self._record_counter[stream.name] += 1
             stream.cursor.observe(record)
-        test_env = os.getenv("PYTEST_CURRENT_TEST")
-        if test_env and "test_concurrent_declarative_source.py" in test_env:
-            self._logger.info(f"Processing and emitting: {message.__dict__}")
         yield message
         yield from self._message_repository.consume_queue()
 

airbyte_cdk/sources/concurrent_source/concurrent_source.py

@@ -3,6 +3,7 @@
 #
 import concurrent
 import logging
+import os
 from queue import Queue
 from typing import Iterable, Iterator, List, Optional
 
@@ -143,6 +144,9 @@ def _consume_from_queue(
         concurrent_stream_processor: ConcurrentReadProcessor,
     ) -> Iterable[AirbyteMessage]:
         while airbyte_message_or_record_or_exception := queue.get():
+            test_env = os.getenv("PYTEST_CURRENT_TEST")
+            if test_env and "test_concurrent_declarative_source.py" in test_env:
+                self._logger.info(f"Processing and emitting: {airbyte_message_or_record_or_exception.__dict__}")
             yield from self._handle_item(
                 airbyte_message_or_record_or_exception,
                 concurrent_stream_processor,