feat(source-s3): migrate to FileBasedConcurrentCursor + bump CDK to prerelease with state throttle

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

@@ -10,7 +10,7 @@ data:
   connectorSubtype: file
   connectorType: source
   definitionId: 69589781-7828-43c5-9f63-8925b1c1ccc2
-  dockerImageTag: 4.15.4
+  dockerImageTag: 4.15.5
   dockerRepository: airbyte/source-s3
   documentationUrl: https://docs.airbyte.com/integrations/sources/s3
   externalDocumentationUrls:

airbyte-integrations/connectors/source-s3/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "4.15.4"
+version = "4.15.5"
 name = "source-s3"
 description = "Source implementation for S3."
 authors = [ "Airbyte <contact@airbyte.io>",]
@@ -22,7 +22,7 @@ wcmatch = "^10.0"
 dill = "^0.3.4"
 transformers = "^4.38.2"
 urllib3 = "<2"
-airbyte-cdk = {extras = ["file-based"], version = "^7.0.0"}
+airbyte-cdk = {version = "7.19.2.post4.dev26418171295", extras = ["file-based"], allow-prereleases = true}
 pendulum = "^3.0.0"
 psutil = "^7.0.0"
 numpy = ">=2.2.6,<3"

airbyte-integrations/connectors/source-s3/source_s3/run.py

@@ -3,8 +3,34 @@
 #
 from __future__ import annotations
 
+
+def _cap_malloc_arenas() -> None:
+    """
+    Cap glibc malloc arenas to keep peak RSS low under the concurrent file-based
+    read path. With the default `8 x N_CPUs` arenas, the source pod can pin 1+ GB
+    of allocator overhead that is never returned to the OS, which combined with
+    the in-flight working set pushes the pod over its 2 Gi memory limit on
+    streams with many or large files (oncall #12663).
+
+    glibc-only. No-op on macOS or musl-libc images; falls back to default
+    behavior, which is correct for those platforms.
+    """
+    try:
+        import ctypes
+
+        libc = ctypes.CDLL("libc.so.6", use_errno=True)
+        # M_ARENA_MAX = -8 from glibc's malloc.h. Equivalent to setting the
+        # MALLOC_ARENA_MAX environment variable, but applied at process start
+        # so it is scoped to this connector only.
+        libc.mallopt(-8, 2)
+    except (OSError, AttributeError):
+        pass
+
+
+_cap_malloc_arenas()
+
 from source_s3.v4 import SourceS3
 
 
 def run() -> None:
-    SourceS3.launch()
+    SourceS3.launch()

airbyte-integrations/connectors/source-s3/source_s3/v4/__init__.py

@@ -3,9 +3,8 @@
 #
 
 from .config import Config
-from .cursor import Cursor
 from .legacy_config_transformer import LegacyConfigTransformer
 from .source import SourceS3
 from .stream_reader import SourceS3StreamReader
 
-__all__ = ["Config", "Cursor", "LegacyConfigTransformer", "SourceS3", "SourceS3StreamReader"]
+__all__ = ["Config", "LegacyConfigTransformer", "SourceS3", "SourceS3StreamReader"]

airbyte-integrations/connectors/source-s3/source_s3/v4/config.py

@@ -92,6 +92,20 @@ def documentation_url(cls) -> AnyUrl:
         default="use_records_transfer",
     )
 
+    concurrency_level: Optional[int] = Field(
+        title="Concurrency Level",
+        description=(
+            "Maximum number of partition reader threads source-s3 uses to read files in parallel. "
+            "Higher values increase throughput but also increase peak memory of the source pod. "
+            "Leave empty to use the CDK default (100). Lower this (e.g. 20-50) if the source pod hits "
+            "its memory limit on streams with many or large files."
+        ),
+        default=None,
+        gt=0,
+        order=7,
+        group="advanced",
+    )
+
     @root_validator
     def validate_optional_args(cls, values):
         aws_access_key_id = values.get("aws_access_key_id")