Fix: source-notion depth leak (#69780)

Co-authored-by: Alfredo Garcia <[email protected]>

Author: eytan-avisror

airbyte-integrations/connectors/source-notion/components.py

@@ -85,6 +85,7 @@ def read_records(
                     cursor_slice=stream_slice.cursor_slice,
                 )
                 yield from self.read_records(records_schema, child_stream_slice)
+                self.current_block_depth -= 1
 
             if "parent" in stream_data:
                 stream_data["parent"]["sequence_number"] = sequence_number

airbyte-integrations/connectors/source-notion/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 6e00b415-b02e-4160-bf02-58176a0ae687
-  dockerImageTag: 3.3.6
+  dockerImageTag: 3.3.7
   dockerRepository: airbyte/source-notion
   documentationUrl: https://docs.airbyte.com/integrations/sources/notion
   externalDocumentationUrls:

airbyte-integrations/connectors/source-notion/unit_tests/test_components.py

@@ -1,11 +1,13 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
 from airbyte_cdk.models import ConfiguredAirbyteCatalogSerializer
+from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
+from airbyte_cdk.sources.declarative.types import Record, StreamSlice
 from airbyte_cdk.test.entrypoint_wrapper import read
 from airbyte_cdk.test.state_builder import StateBuilder
 from unit_tests.conftest import get_source
@@ -97,6 +99,65 @@ def test_notion_properties_transformation(components_module):
 ]
 
 
+def test_blocks_retriever_depth_restored_after_children(components_module):
+    """
+    Test to verify that current_block_depth is correctly incremented and decremented during recursive calls.
+
+    This test creates a scenario with two sibling blocks at the same level, both having children:
+    - Block A (has_children=True)
+      - Block A1 (has_children=False)
+    - Block B (has_children=True)
+      - Block B1 (has_children=False)
+
+    The depth counter should properly increment when entering a child level and decrement when
+    returning to the parent level. This ensures sibling blocks are processed at the correct depth.
+    """
+
+    retriever = components_module.BlocksRetriever(
+        name="test",
+        primary_key=["id"],
+        requester=MagicMock(),
+        record_selector=MagicMock(),
+        paginator=MagicMock(),
+        config={},
+        parameters={},
+    )
+    retriever._paginator = MagicMock()
+    depth_tracker = []
+
+    block_a = Record(
+        data={"id": "block-a", "has_children": True}, stream_name="blocks", associated_slice=StreamSlice(partition={}, cursor_slice={})
+    )
+    block_a1 = Record(
+        data={"id": "block-a1", "has_children": False}, stream_name="blocks", associated_slice=StreamSlice(partition={}, cursor_slice={})
+    )
+    block_b = Record(
+        data={"id": "block-b", "has_children": True}, stream_name="blocks", associated_slice=StreamSlice(partition={}, cursor_slice={})
+    )
+    block_b1 = Record(
+        data={"id": "block-b1", "has_children": False}, stream_name="blocks", associated_slice=StreamSlice(partition={}, cursor_slice={})
+    )
+
+    def mock_super_read_records(self, records_schema, stream_slice=None):
+        depth_tracker.append(retriever.current_block_depth)
+
+        if stream_slice is None or not stream_slice.partition:
+            yield block_a
+            yield block_b
+        elif stream_slice.partition.get("block_id") == "block-a":
+            yield block_a1
+        elif stream_slice.partition.get("block_id") == "block-b":
+            yield block_b1
+
+    with patch.object(SimpleRetriever, "read_records", mock_super_read_records):
+        stream_slice = StreamSlice(partition={}, cursor_slice={})
+        results = list(retriever.read_records({}, stream_slice))
+
+    assert len(depth_tracker) == 3, f"Expected 3 depth measurements, got {len(depth_tracker)}: {depth_tracker}"
+    assert depth_tracker == [0, 1, 1], f"Depth should be properly managed. Depth tracker: {depth_tracker}"
+    assert retriever.current_block_depth == 0, f"Final depth should be 0, got {retriever.current_block_depth}"
+
+
 def test_blocks_retriever(requests_mock):
     page_response_data = {
         "object": "list",

docs/integrations/sources/notion.md

@@ -119,6 +119,7 @@ The connector is restricted by Notion [request limits](https://developers.notion
 
 | Version     | Date       | Pull Request                                             | Subject                                                                                                                                                                |
 |:------------|:-----------|:---------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.3.7 | 2025-11-24 | [69375](https://github.com/airbytehq/airbyte/pull/69780) | Fix bug in current_block_depth |
 | 3.3.6 | 2025-11-18 | [69375](https://github.com/airbytehq/airbyte/pull/69375) | Update dependencies |
 | 3.3.5 | 2025-10-29 | [68750](https://github.com/airbytehq/airbyte/pull/68750) | Update dependencies |
 | 3.3.4 | 2025-10-21 | [68399](https://github.com/airbytehq/airbyte/pull/68399) | Update dependencies |