DO NOT MERGE (test only): pin prerelease SDM base image to validate OC-12975 fix in Cloud

airbyte-integrations/connectors/source-stripe/manifest.yaml

@@ -1468,6 +1468,8 @@ definitions:
           requester:
             $ref: "#/definitions/base_requester"
             path: events
+            request_parameters:
+              types[]: '{{["invoice.created", "invoice.deleted", "invoice.updated"]}}'
           record_selector:
             type: RecordSelector
             extractor:
@@ -1482,12 +1484,27 @@ definitions:
                   - lines
                   - data
                 remain_original_record: true
+                truncation_indicator_path:
+                  - data
+                  - object
+                  - lines
+                  - has_more
+                truncated_list_retriever:
+                  type: SimpleRetriever
+                  requester:
+                    $ref: "#/definitions/base_requester"
+                    path: invoices/{{ stream_slice['parent_record']['data']['object']['id'] }}/lines
+                  record_selector:
+                    type: RecordSelector
+                    extractor:
+                      type: DpathExtractor
+                      field_path:
+                        - data
+                  paginator:
+                    $ref: "#/definitions/base_paginator"
             transform_before_filtering: true
           paginator:
             $ref: "#/definitions/base_paginator"
-          $parameters:
-            request_parameters:
-              types[]: '{{["invoice.created", "invoice.deleted", "invoice.updated"]}}'
         schema_loader:
           type: InlineSchemaLoader
           schema:

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

@@ -6,11 +6,11 @@ data:
     hosts:
       - api.stripe.com
   connectorBuildOptions:
-    baseImage: docker.io/airbyte/source-declarative-manifest:7.25.1@sha256:cdbd1578932d121f2852c065fbabebc49ccbf00078606dc503be5e4c66b373df
+    baseImage: docker.io/airbyte/source-declarative-manifest:7.28.2.post5.dev33128883500@sha256:ec8060b750ce67b0cb0b04dc56152d8afb0c3ef21e41bc5a97b1972f00b74945
   connectorSubtype: api
   connectorType: source
   definitionId: e094cb9a-26de-4645-8761-65c0c425d1de
-  dockerImageTag: 6.0.15
+  dockerImageTag: 6.0.16
   dockerRepository: airbyte/source-stripe
   documentationUrl: https://docs.airbyte.com/integrations/sources/stripe
   erdUrl: https://dbdocs.io/airbyteio/source-stripe?view=relationships

airbyte-integrations/connectors/source-stripe/unit_tests/integration/request_builder.py

@@ -42,6 +42,10 @@ def events_endpoint(cls, account_id: str, client_secret: str) -> "StripeRequestB
     def external_accounts_endpoint(cls, account_id: str, client_secret: str) -> "StripeRequestBuilder":
         return cls(f"accounts/{account_id}/external_accounts", account_id, client_secret)
 
+    @classmethod
+    def invoice_lines_endpoint(cls, invoice_id: str, account_id: str, client_secret: str) -> "StripeRequestBuilder":
+        return cls(f"invoices/{invoice_id}/lines", account_id, client_secret)
+
     @classmethod
     def issuing_authorizations_endpoint(cls, account_id: str, client_secret: str) -> "StripeRequestBuilder":
         return cls("issuing/authorizations", account_id, client_secret)

airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_invoice_line_items.py

@@ -0,0 +1,185 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List
+from unittest import TestCase
+
+import freezegun
+from unit_tests.conftest import get_source
+
+from airbyte_cdk.models import ConfiguredAirbyteCatalog, SyncMode
+from airbyte_cdk.test.catalog_builder import CatalogBuilder
+from airbyte_cdk.test.entrypoint_wrapper import read
+from airbyte_cdk.test.mock_http import HttpMocker
+from airbyte_cdk.test.mock_http.response_builder import (
+    FieldPath,
+    HttpResponseBuilder,
+    NestedPath,
+    RecordBuilder,
+    create_record_builder,
+    create_response_builder,
+    find_template,
+)
+from airbyte_cdk.test.state_builder import StateBuilder
+from integration.config import ConfigBuilder
+from integration.pagination import StripePaginationStrategy
+from integration.request_builder import StripeRequestBuilder
+
+
+_STREAM_NAME = "invoice_line_items"
+_ACCOUNT_ID = "acct_1G9HZLIEn49ers"
+_CLIENT_SECRET = "ConfigBuilder default client secret"
+_INVOICE_ID = "in_1K9GK0EcXtiJtvvhSo2LvGqT"
+_NOW = datetime.now(timezone.utc)
+_START_DATE = _NOW - timedelta(days=75)
+_STATE_DATE = _NOW - timedelta(days=10)
+
+
+def _config() -> ConfigBuilder:
+    return ConfigBuilder().with_account_id(_ACCOUNT_ID).with_client_secret(_CLIENT_SECRET).with_slice_range_in_days(365)
+
+
+def _create_catalog(sync_mode: SyncMode = SyncMode.incremental) -> ConfiguredAirbyteCatalog:
+    return CatalogBuilder().with_stream(name=_STREAM_NAME, sync_mode=sync_mode).build()
+
+
+def _events_request() -> StripeRequestBuilder:
+    return StripeRequestBuilder.events_endpoint(_ACCOUNT_ID, _CLIENT_SECRET)
+
+
+def _invoice_lines_request() -> StripeRequestBuilder:
+    return StripeRequestBuilder.invoice_lines_endpoint(_INVOICE_ID, _ACCOUNT_ID, _CLIENT_SECRET)
+
+
+def _line_item(line_item_id: str) -> Dict[str, Any]:
+    return {
+        "id": line_item_id,
+        "object": "line_item",
+        "amount": 1000,
+        "currency": "usd",
+        "invoice": _INVOICE_ID,
+        "period": {"start": int(_STATE_DATE.timestamp()), "end": int(_STATE_DATE.timestamp())},
+        "type": "invoiceitem",
+    }
+
+
+def _invoice(embedded_lines: List[Dict[str, Any]], has_more: bool, total_count: int) -> Dict[str, Any]:
+    return {
+        "id": _INVOICE_ID,
+        "object": "invoice",
+        "created": int(_STATE_DATE.timestamp()) + 1,
+        "lines": {
+            "object": "list",
+            "data": embedded_lines,
+            "has_more": has_more,
+            "total_count": total_count,
+            "url": f"/v1/invoices/{_INVOICE_ID}/lines",
+        },
+    }
+
+
+def _event_record() -> RecordBuilder:
+    return create_record_builder(
+        find_template("events", __file__),
+        FieldPath("data"),
+        record_id_path=FieldPath("id"),
+        record_cursor_path=FieldPath("created"),
+    )
+
+
+def _events_response() -> HttpResponseBuilder:
+    return create_response_builder(find_template("events", __file__), FieldPath("data"), pagination_strategy=StripePaginationStrategy())
+
+
+def _invoice_created_event(invoice: Dict[str, Any]) -> RecordBuilder:
+    return (
+        _event_record()
+        .with_field(FieldPath("type"), "invoice.created")
+        .with_field(FieldPath("created"), int(_STATE_DATE.timestamp()) + 1)
+        .with_field(NestedPath(["data", "object"]), invoice)
+    )
+
+
+def _invoice_line_record(line_item_id: str) -> RecordBuilder:
+    return create_record_builder(
+        find_template("invoice_lines", __file__),
+        FieldPath("data"),
+        record_id_path=FieldPath("id"),
+    ).with_id(line_item_id)
+
+
+def _invoice_lines_response() -> HttpResponseBuilder:
+    return create_response_builder(
+        response_template=find_template("invoice_lines", __file__),
+        records_path=FieldPath("data"),
+        pagination_strategy=StripePaginationStrategy(),
+    )
+
+
+def _read_incremental(http_mocker: HttpMocker) -> Any:
+    config = _config().with_start_date(_START_DATE).build()
+    state = StateBuilder().with_stream_state(_STREAM_NAME, {"invoice_updated": int(_STATE_DATE.timestamp())}).build()
+    source = get_source(config=config, state=state)
+    return read(source, config=config, catalog=_create_catalog(), state=state)
+
+
+@freezegun.freeze_time(_NOW.isoformat())
+class InvoiceLineItemsIncrementalTest(TestCase):
+    @HttpMocker()
+    def test_given_truncated_embedded_lines_when_read_then_fetch_all_lines_from_invoice_endpoint(self, http_mocker: HttpMocker) -> None:
+        all_line_ids = [f"il_{index}" for index in range(15)]
+        embedded_lines = [_line_item(line_item_id) for line_item_id in all_line_ids[:10]]
+        http_mocker.get(
+            _events_request().with_any_query_params().build(),
+            _events_response().with_record(_invoice_created_event(_invoice(embedded_lines, has_more=True, total_count=15))).build(),
+        )
+        first_page = _invoice_lines_response()
+        for line_item_id in all_line_ids[:10]:
+            first_page = first_page.with_record(_invoice_line_record(line_item_id))
+        second_page = _invoice_lines_response()
+        for line_item_id in all_line_ids[10:]:
+            second_page = second_page.with_record(_invoice_line_record(line_item_id))
+        http_mocker.get(
+            _invoice_lines_request().with_limit(100).build(),
+            first_page.with_pagination().build(),
+        )
+        http_mocker.get(
+            _invoice_lines_request().with_limit(100).with_starting_after("il_9").build(),
+            second_page.build(),
+        )
+
+        output = _read_incremental(http_mocker)
+
+        assert len(output.records) == 15
+        assert sorted(record.record.data["id"] for record in output.records) == sorted(all_line_ids)
+        assert all(record.record.data["invoice_id"] == _INVOICE_ID for record in output.records)
+
+    @HttpMocker()
+    def test_given_all_lines_embedded_when_read_then_extract_all_lines_without_extra_request(self, http_mocker: HttpMocker) -> None:
+        all_line_ids = [f"il_{index}" for index in range(15)]
+        embedded_lines = [_line_item(line_item_id) for line_item_id in all_line_ids]
+        http_mocker.get(
+            _events_request().with_any_query_params().build(),
+            _events_response().with_record(_invoice_created_event(_invoice(embedded_lines, has_more=False, total_count=15))).build(),
+        )
+
+        output = _read_incremental(http_mocker)
+
+        assert len(output.records) == 15
+        assert sorted(record.record.data["id"] for record in output.records) == sorted(all_line_ids)
+
+    @HttpMocker()
+    def test_given_few_embedded_lines_when_read_then_no_request_to_invoice_lines_endpoint(self, http_mocker: HttpMocker) -> None:
+        all_line_ids = [f"il_{index}" for index in range(10)]
+        embedded_lines = [_line_item(line_item_id) for line_item_id in all_line_ids]
+        http_mocker.get(
+            _events_request().with_any_query_params().build(),
+            _events_response().with_record(_invoice_created_event(_invoice(embedded_lines, has_more=False, total_count=10))).build(),
+        )
+
+        output = _read_incremental(http_mocker)
+
+        assert len(output.records) == 10
+        assert sorted(record.record.data["id"] for record in output.records) == sorted(all_line_ids)

airbyte-integrations/connectors/source-stripe/unit_tests/resource/http/response/invoice_lines.json

@@ -0,0 +1,23 @@
+{
+  "object": "list",
+  "data": [
+    {
+      "id": "il_1OEiWvEcXtiJtvvh0uxUZGRW",
+      "object": "line_item",
+      "amount": 1000,
+      "currency": "usd",
+      "description": "line item description",
+      "discountable": true,
+      "livemode": false,
+      "period": {
+        "end": 1700529213,
+        "start": 1700529213
+      },
+      "proration": false,
+      "quantity": 1,
+      "type": "invoiceitem"
+    }
+  ],
+  "has_more": false,
+  "url": "/v1/invoices/in_1OEiWvEcXtiJtvvh/lines"
+}

docs/integrations/sources/stripe.md

@@ -317,6 +317,7 @@ If you use Airbyte Cloud and your organization restricts access to specific IPs,
 
 | Version     | Date       | Pull Request                                                 | Subject                                                                                                                                                                                                                       |
 |:------------|:-----------|:-------------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 6.0.16 | 2026-08-26 | [85087](https://github.com/airbytehq/airbyte/pull/85087) | Fix truncated `invoice_line_items` on incremental syncs: fetch the complete line item list from `invoices/{id}/lines` when the event payload embeds only the first page. Existing destination data for invoices with more than 10 line items requires a Full Refresh to repair. |
 | 6.0.15 | 2026-08-18 | [84768](https://github.com/airbytehq/airbyte/pull/84768) | Update dependencies |
 | 6.0.14 | 2026-08-17 | [84355](https://github.com/airbytehq/airbyte/pull/84355) | Update events now win same-second cursor ties with creation events so the newer payload is kept at the destination. |
 | 6.0.13 | 2026-08-11 | [84134](https://github.com/airbytehq/airbyte/pull/84134) | Update dependencies |