File-based CDK: enqueue AirbyteMessage of type record instead of send…

…ing to the message repository (airbytehq#35318)
ollionorg · Feb 21, 2024 · d8843be · d8843be
1 parent 739405c
commit d8843be
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/file_based/stream/concurrent/adapters.py b/airbyte-cdk/python/airbyte_cdk/sources/file_based/stream/concurrent/adapters.py
@@ -243,6 +243,9 @@ def read(self) -> Iterable[Record]:
                     data_to_return = dict(record_data)
                     self._stream.transformer.transform(data_to_return, self._stream.get_json_schema())
                     yield Record(data_to_return, self.stream_name())
+                elif isinstance(record_data, AirbyteMessage) and record_data.type == Type.RECORD:
+                    # `AirbyteMessage`s of type `Record` should also be yielded so they are enqueued
+                    yield Record(record_data.record.data, self.stream_name())
                 else:
                     self._message_repository.emit_message(record_data)
         except Exception as e: