File-based CDK: enqueue AirbyteMessage of type record instead of sending to the message repository (#35318)

Author: clnoll

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: