From a17d00d7bd60191ef050fc4ac74e707918674d2a Mon Sep 17 00:00:00 2001 From: pnilan Date: Mon, 9 Mar 2026 15:56:17 -0700 Subject: [PATCH] DO NOT MERGE: add intentional memory leak to test graceful OOM shutdown Duplicates each consumed record 500x into a growing list to rapidly inflate memory and trigger the MemoryMonitor warning/critical thresholds. --- airbyte_cdk/entrypoint.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py index ad290ebce3..76fac50781 100644 --- a/airbyte_cdk/entrypoint.py +++ b/airbyte_cdk/entrypoint.py @@ -279,9 +279,16 @@ def read( # The Airbyte protocol dictates that counts be expressed as float/double to better protect against integer overflows stream_message_counter: DefaultDict[HashableStreamDescriptor, float] = defaultdict(float) + + # DO NOT MERGE: intentional memory leak to test graceful OOM shutdown. + # Each consumed record is duplicated 500x into a growing list to inflate memory quickly. + _leaked_records: list[Any] = [] + try: for message in self.source.read(self.logger, config, catalog, state): self._memory_monitor.check_memory_usage() + if message.type == Type.RECORD: + _leaked_records.extend([message.record] * 500) yield self.handle_record_counts(message, stream_message_counter) finally: for message in self._emit_queued_messages(self.source):