[DOP-24124] Add Kafka GSSAPI support

Author: dolfinus

.env.docker

@@ -27,8 +27,8 @@ DATA_RENTGEN__LOGGING__PRESET=colored
 DATA_RENTGEN__SERVER__DEBUG=false
 
 # See Backend -> Consumer -> Configuration documentation
-DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=broker:9092
-DATA_RENTGEN__KAFKA__SECURITY__TYPE=scram-sha256
+DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS='["broker:9092"]'
+DATA_RENTGEN__KAFKA__SECURITY__TYPE=SCRAM-SHA-256
 DATA_RENTGEN__KAFKA__SECURITY__USER=data_rentgen
 DATA_RENTGEN__KAFKA__SECURITY__PASSWORD=changeme
 DATA_RENTGEN__KAFKA__COMPRESSION=zstd

.env.local

@@ -2,8 +2,8 @@ export DATA_RENTGEN__LOGGING__PRESET=colored
 
 export DATA_RENTGEN__DATABASE__URL=postgresql+asyncpg://data_rentgen:changeme@localhost:5432/data_rentgen
 
-export DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=localhost:9093
-export DATA_RENTGEN__KAFKA__SECURITY__TYPE=scram-sha256
+export DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=["localhost:9093"]
+export DATA_RENTGEN__KAFKA__SECURITY__TYPE=SCRAM-SHA-256
 export DATA_RENTGEN__KAFKA__SECURITY__USER=data_rentgen
 export DATA_RENTGEN__KAFKA__SECURITY__PASSWORD=changeme
 export DATA_RENTGEN__KAFKA__COMPRESSION=zstd
@@ -36,9 +36,9 @@ export DATA_RENTGEN__AUTH__ACCESS_TOKEN__SECRET_KEY=secret
 
 # Cors
 export DATA_RENTGEN__SERVER__CORS__ENABLED=True
-export 'DATA_RENTGEN__SERVER__CORS__ALLOW_ORIGINS=["http://localhost:3000"]'
-export DATA_RENTGEN__SERVER__CORS__ALLOW_CREDENTIALS=true
-export 'DATA_RENTGEN__SERVER__CORS__ALLOW_METHODS=["*"]'
-export 'DATA_RENTGEN__SERVER__CORS__ALLOW_HEADERS=["*"]'
-export 'DATA_RENTGEN__SERVER__CORS__EXPOSE_HEADERS=["X-Request-ID", "Location", "Access-Control-Allow-Credentials"]'
+export DATA_RENTGEN__SERVER__CORS__ALLOW_ORIGINS=["http://localhost:3000"]
+export DATA_RENTGEN__SERVER__CORS__ALLOW_CREDENTIALS=True
+export DATA_RENTGEN__SERVER__CORS__ALLOW_METHODS=["*"]
+export DATA_RENTGEN__SERVER__CORS__ALLOW_HEADERS=["*"]
+export DATA_RENTGEN__SERVER__CORS__EXPOSE_HEADERS=["X-Request-ID", "Location", "Access-Control-Allow-Credentials"]
 

.github/workflows/codeql-analysis.yml

@@ -29,6 +29,11 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v4
 
+      - name: Install system packages
+        run: |
+          sudo apt update
+          sudo apt-get install --no-install-recommends -y libkrb5-dev krb5-user gcc
+
       - name: Set up Python ${{ env.DEFAULT_PYTHON }}
         uses: actions/setup-python@v5
         with:

.github/workflows/test.yml

@@ -24,6 +24,11 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
 
+      - name: Install system packages
+        run: |
+          sudo apt update
+          sudo apt-get install --no-install-recommends -y libkrb5-dev krb5-user gcc
+
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
@@ -97,7 +102,7 @@ jobs:
         run: python -m pip install --upgrade pip setuptools wheel
 
       - name: Install dependencies
-        run: pip install -I coverage pytest
+        run: pip install -I coverage
 
       - name: Download all raw coverage data
         uses: actions/download-artifact@v4

.readthedocs.yaml

@@ -3,9 +3,12 @@ version: 2
 build:
   os: ubuntu-22.04
   apt_packages:
+    - autoconf
     - make
+    - gcc
+    - libkrb5-dev
   tools:
-    python: '3.12'
+    python: '3.13'
   jobs:
     post_checkout:
       - git fetch --unshallow || true

codecov.yml

@@ -2,5 +2,5 @@ coverage:
   status:
     project:
       default:
-        target: 93%
+        target: 90%
         threshold: 1%

data_rentgen/consumer/__init__.py

@@ -2,15 +2,17 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
+from contextlib import asynccontextmanager
 
+import anyio
 from fast_depends import dependency_provider
-from faststream import FastStream
+from faststream import ContextRepo, FastStream
+from faststream._compat import ExceptionGroup
 from faststream.kafka import KafkaBroker
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import data_rentgen
 from data_rentgen.consumer.settings import ConsumerApplicationSettings
-from data_rentgen.consumer.settings.security import get_broker_security
 from data_rentgen.consumer.subscribers import runs_events_subscriber
 from data_rentgen.db.factory import create_session_factory
 from data_rentgen.logging.setup_logging import setup_logging
@@ -21,10 +23,11 @@
 def broker_factory(settings: ConsumerApplicationSettings) -> KafkaBroker:
     broker = KafkaBroker(
         bootstrap_servers=settings.kafka.bootstrap_servers,
-        security=get_broker_security(settings.kafka.security),
+        security=settings.kafka.security.to_security(),
         compression_type=settings.kafka.compression.value if settings.kafka.compression else None,
         client_id=f"data-rentgen-{data_rentgen.__version__}",
         logger=logger,
+        **settings.kafka.security.extra_broker_kwargs(),
     )
 
     # register subscribers using settings
@@ -41,8 +44,24 @@ def broker_factory(settings: ConsumerApplicationSettings) -> KafkaBroker:
 
 
 def application_factory(settings: ConsumerApplicationSettings) -> FastStream:
+    @asynccontextmanager
+    async def security_lifespan(context: ContextRepo):
+        try:
+            async with anyio.create_task_group() as tg:
+                await settings.kafka.security.initialize()
+                tg.start_soon(settings.kafka.security.refresh)
+
+                yield
+
+                await settings.kafka.security.destroy()
+                tg.cancel_scope.cancel()
+        except ExceptionGroup as e:
+            for exception in e.exceptions:
+                raise exception from None
+
     return FastStream(
         broker=broker_factory(settings),
+        lifespan=security_lifespan,
         title="Data.Rentgen",
         description="Data.Rentgen is a nextgen DataLineage service",
         version=data_rentgen.__version__,

data_rentgen/consumer/settings/__init__.py

@@ -31,8 +31,8 @@ class ConsumerApplicationSettings(BaseSettings):
         # same as settings.database.url = "postgresql+asyncpg://postgres:postgres@localhost:5432/data_rentgen"
         DATA_RENTGEN__DATABASE__URL=postgresql+asyncpg://postgres:postgres@localhost:5432/data_rentgen
 
-        # same as settings.kafka.bootstrap_servers = "postgresql+asyncpg://postgres:postgres@localhost:5432/data_rentgen"
-        DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=postgresql+asyncpg://postgres:postgres@localhost:5432/data_rentgen
+        # same as settings.kafka.bootstrap_servers = ["kafka:9092"]
+        DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=["kafka:9092"]
 
         # same as settings.logging.preset = "json"
         DATA_RENTGEN__LOGGING__PRESET=json

data_rentgen/consumer/settings/kafka.py

@@ -6,7 +6,7 @@
 
 from pydantic import BaseModel, Field
 
-from data_rentgen.consumer.settings.security import KafkaSecuritySettings
+from data_rentgen.consumer.settings.security import KafkaSecurityAnonymousSettings, KafkaSecuritySettings
 
 
 class KafkaCompression(str, Enum):
@@ -30,18 +30,18 @@ class KafkaSettings(BaseModel):
 
     .. code-block:: bash
 
-        DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=localhost:9092
-        DATA_RENTGEN__KAFKA__SECURITY__TYPE=scram-256
+        DATA_RENTGEN__KAFKA__BOOTSTRAP_SERVERS=["localhost:9092"]
+        DATA_RENTGEN__KAFKA__SECURITY__TYPE=SCRAM-SHA-256
         DATA_RENTGEN__KAFKA__REQUEST_TIMEOUT_MS=5000
         DATA_RENTGEN__KAFKA__CONNECTIONS_MAX_IDLE_MS=540000
     """
 
-    bootstrap_servers: str = Field(
+    bootstrap_servers: list[str] = Field(
         description="List of Kafka bootstrap servers.",
         min_length=1,
     )
     security: KafkaSecuritySettings = Field(
-        default_factory=KafkaSecuritySettings,
+        default_factory=KafkaSecurityAnonymousSettings,
         description="Kafka security settings.",
     )
     compression: KafkaCompression | None = Field(