feat: add bounded kafka ingestion pipeline

.env.sample

@@ -4,7 +4,11 @@
 
 REDPANDA_VERSION=v26.2.3
 
-KAFKA_BOOTSTRAP_SERVERS=redpanda:9092
+# Used by Python applications running on the host.
+KAFKA_BOOTSTRAP_SERVERS=localhost:19092
+
+# Used by containers communicating inside the Docker network.
+KAFKA_INTERNAL_BOOTSTRAP_SERVERS=redpanda:9092
 
 # Raw CDR topic
 KAFKA_RAW_TOPIC=cdr.raw
@@ -22,6 +26,20 @@ KAFKA_DLQ_TOPIC_RETENTION_MS=604800000
 KAFKA_CONSUMER_GROUP=cdr-processing-workers
 
 
+# =========================
+# API ingestion
+# =========================
+
+API_QUEUE_SIZE=10000
+
+KAFKA_DELIVERY_TIMEOUT_SECONDS=4
+KAFKA_MESSAGE_TIMEOUT_MS=3000
+
+KAFKA_LINGER_MS=5
+KAFKA_BATCH_SIZE=131072
+KAFKA_PRODUCER_MAX_MESSAGES=100000
+
+
 # =========================
 # ClickHouse
 # =========================

README.md

@@ -21,7 +21,9 @@ FastAPI API
     ↓
 Validation
     ↓
-202 Accepted
+Bounded Async Queue
+    ↓
+Redpanda / cdr.raw
 ```
 
 Infrastructure already available:
@@ -35,7 +37,7 @@ ClickHouse
 └── telecom.cdr_records
 ```
 
-Kafka publishing and worker processing are not connected yet.
+worker processing is not connected yet.
 
 ## Current Features
 
@@ -53,16 +55,27 @@ Kafka publishing and worker processing are not connected yet.
 * FastAPI `GET /health`
 * Structured JSON error responses
 * Unit and API integration tests
+* Bounded asynchronous ingestion queue for backpressure
+* Long-lived asynchronous Kafka producer
+* Kafka delivery acknowledgement before `202 Accepted`
+* `429 Too Many Requests` on ingestion queue saturation
+* `503 Service Unavailable` on Kafka delivery failure
 
 ## Project Structure
 
 ```text
 src/core/
+├── config.py
+│
 ├── api/
+│   ├── __init__.py
+│   ├── kafka_publisher.py
 │   ├── main.py
+│   ├── publisher.py
 │   └── routes.py
 │
 └── domain/
+    ├── __init__.py
     ├── errors.py
     ├── models.py
     ├── normalization.py

docker-compose.yml

@@ -56,7 +56,7 @@ services:
         condition: service_healthy
 
     environment:
-      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS}
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS}
 
       KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC}
       KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS}

pyproject.toml

@@ -7,6 +7,8 @@ dependencies = [
     "phonenumbers>=8.13",
     "fastapi>=0.115,<1.0",
     "uvicorn[standard]>=0.30,<1.0",
+    "confluent-kafka>=2.15,<3.0",
+    "pydantic-settings>=2.0,<3.0",
 ]
 
 [project.optional-dependencies]

scripts/sample-valid-cdr.json

@@ -0,0 +1,14 @@
+{
+  "raw_id": "raw-test-001",
+  "src_ip": "10.240.12.1",
+  "device_id": "310410998877665",
+  "caller_num": "0019875551234",
+  "svc": "4G_DATA",
+  "up_bytes": 1048576,
+  "down_bytes": 4194304,
+  "rat": 6,
+  "lac": "4021",
+  "ci": "8812",
+  "epoch_start": 1789900000,
+  "epoch_end": 1789900300
+}

src/core/api/kafka_publisher.py

@@ -0,0 +1,184 @@
+import asyncio
+import json
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from confluent_kafka import KafkaException, Producer
+
+from core.api.publisher import (
+    IngestionQueueFull,
+    KafkaPublishError,
+)
+from core.config import Settings
+
+
+@dataclass(slots=True)
+class PendingMessage:
+    key: str
+    value: bytes
+    result: asyncio.Future[None]
+
+
+class KafkaPublisher:
+    def __init__(self, settings: Settings) -> None:
+        self._settings = settings
+
+        self._queue: asyncio.Queue[PendingMessage] = asyncio.Queue(
+            maxsize=settings.api_queue_size,
+        )
+
+        self._producer = Producer(
+            {
+                "bootstrap.servers": settings.kafka_bootstrap_servers,
+
+                "client.id": "cdr-ingestion-api",
+
+                "acks": "all",
+                "enable.idempotence": True,
+
+                "linger.ms": settings.kafka_linger_ms,
+                "batch.size": settings.kafka_batch_size,
+
+                "compression.type": "lz4",
+
+                "queue.buffering.max.messages":
+                    settings.kafka_producer_max_messages,
+
+                "message.timeout.ms":
+                    settings.kafka_message_timeout_ms,
+            }
+        )
+
+        self._task: asyncio.Task[None] | None = None
+        self._stopping = False
+
+    async def start(self) -> None:
+        if self._task is not None:
+            return
+
+        self._stopping = False
+
+        self._task = asyncio.create_task(
+            self._run(),
+            name="kafka-publisher",
+        )
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        loop = asyncio.get_running_loop()
+
+        result: asyncio.Future[None] = loop.create_future()
+
+        value = json.dumps(
+            payload,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        ).encode("utf-8")
+
+        message = PendingMessage(
+            key=key,
+            value=value,
+            result=result,
+        )
+
+        try:
+            self._queue.put_nowait(message)
+
+        except asyncio.QueueFull as exc:
+            raise IngestionQueueFull(
+                "Ingestion queue is full."
+            ) from exc
+
+        try:
+            await asyncio.wait_for(
+                result,
+                timeout=self._settings.kafka_delivery_timeout_seconds,
+            )
+
+        except TimeoutError as exc:
+            raise KafkaPublishError(
+                "Kafka delivery acknowledgement timed out."
+            ) from exc
+
+    def _send_to_kafka(
+        self,
+        pending: PendingMessage,
+    ) -> None:
+        def on_delivery(error, message) -> None:
+            if pending.result.done():
+                return
+
+            if error is not None:
+                pending.result.set_exception(
+                    KafkaPublishError(str(error))
+                )
+                return
+
+            pending.result.set_result(None)
+
+        try:
+            self._producer.produce(
+                topic=self._settings.kafka_raw_topic,
+                key=pending.key,
+                value=pending.value,
+                on_delivery=on_delivery,
+            )
+
+        except BufferError as exc:
+            if not pending.result.done():
+                pending.result.set_exception(
+                    KafkaPublishError(
+                        "Kafka producer queue is full."
+                    )
+                )
+
+        except KafkaException as exc:
+            if not pending.result.done():
+                pending.result.set_exception(
+                    KafkaPublishError(str(exc))
+                )
+
+    async def _run(self) -> None:
+        while (
+            not self._stopping
+            or not self._queue.empty()
+            or len(self._producer) > 0
+        ):
+            pending: PendingMessage | None = None
+
+            try:
+                pending = await asyncio.wait_for(
+                    self._queue.get(),
+                    timeout=0.05,
+                )
+
+            except TimeoutError:
+                pass
+
+            if pending is not None:
+                self._send_to_kafka(pending)
+                self._queue.task_done()
+
+            # Triggers Kafka delivery callbacks.
+            self._producer.poll(0)
+
+            # Give the event loop a chance to handle HTTP requests.
+            await asyncio.sleep(0)
+
+    async def stop(self) -> None:
+        if self._task is None:
+            return
+
+        self._stopping = True
+
+        await self._task
+
+        self._task = None
+
+    @property
+    def queue_size(self) -> int:
+        return self._queue.qsize()

src/core/api/main.py

@@ -1,22 +1,53 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 
+from core.api.kafka_publisher import KafkaPublisher
+from core.api.publisher import Publisher
 from core.api.routes import router
+from core.config import get_settings
 
 
+def create_app(
+    publisher: Publisher | None = None,
+) -> FastAPI:
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        active_publisher = publisher
+
+        if active_publisher is None:
+            active_publisher = KafkaPublisher(
+                get_settings()
+            )
+
+        app.state.publisher = active_publisher
+
+        await active_publisher.start()
+
+        try:
+            yield
+
+        finally:
+            await active_publisher.stop()
+
     app = FastAPI(
         title="Telecom CDR Ingestion API",
         version="0.1.0",
+        lifespan=lifespan,
     )
 
-
     app.include_router(
         router,
         prefix="/api/v1",
     )
 
-
     @app.get("/health")
     async def health() -> dict[str, str]:
         return {
             "status": "ok",
         }
+
+    return app
+
+
+app = create_app()

src/core/api/publisher.py

@@ -0,0 +1,25 @@
+from collections.abc import Mapping
+from typing import Any, Protocol
+
+
+class IngestionQueueFull(Exception):
+    pass
+
+
+class KafkaPublishError(Exception):
+    pass
+
+
+class Publisher(Protocol):
+    async def start(self) -> None:
+        ...
+
+    async def stop(self) -> None:
+        ...
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        ...

src/core/api/routes.py

@@ -3,6 +3,7 @@ import json
 from fastapi import APIRouter, Request, status
 from fastapi.responses import JSONResponse
 
+from core.api.publisher import IngestionQueueFull, KafkaPublishError
 from core.domain.errors import CDRProcessingError
 from core.domain.validation import parse_raw_cdr
 
@@ -49,6 +50,30 @@ async def ingest_raw_cdr(request: Request) -> JSONResponse:
             },
         )
 
+    try:
+        await request.app.state.publisher.publish(
+            payload,
+            key=raw_cdr.raw_id,
+        )
+
+    except IngestionQueueFull:
+        return JSONResponse(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            content={
+                "error": "backpressure",
+                "detail": "Ingestion queue is full.",
+            },
+        )
+
+    except KafkaPublishError as exc:
+        return JSONResponse(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            content={
+                "error": "kafka_unavailable",
+                "detail": str(exc),
+            },
+        )
+
     return JSONResponse(
         status_code=status.HTTP_202_ACCEPTED,
         content={

src/core/config.py

@@ -0,0 +1,28 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    kafka_bootstrap_servers: str = "localhost:19092"
+    kafka_raw_topic: str = "cdr.raw"
+
+    api_queue_size: int = 10_000
+
+    kafka_delivery_timeout_seconds: float = 4.0
+    kafka_message_timeout_ms: int = 3_000
+
+    kafka_linger_ms: int = 5
+    kafka_batch_size: int = 131_072
+    kafka_producer_max_messages: int = 100_000
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

tests/integration/test_ingestion_api.py

@@ -1,9 +1,58 @@
+from collections.abc import Mapping
+from typing import Any
+
+import pytest
 from fastapi.testclient import TestClient
 
-from core.api.main import app
+from core.api.main import create_app
+from core.api.publisher import (
+    IngestionQueueFull,
+    KafkaPublishError,
+)
 
 
-client = TestClient(app)
+class FakePublisher:
+    def __init__(self) -> None:
+        self.messages: list[
+            tuple[str, Mapping[str, Any]]
+        ] = []
+
+        self.failure: Exception | None = None
+
+    async def start(self) -> None:
+        pass
+
+    async def stop(self) -> None:
+        pass
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        if self.failure is not None:
+            raise self.failure
+
+        self.messages.append(
+            (key, payload)
+        )
+
+
+@pytest.fixture
+def publisher() -> FakePublisher:
+    return FakePublisher()
+
+
+@pytest.fixture
+def client(
+    publisher: FakePublisher,
+):
+    app = create_app(
+        publisher=publisher,
+    )
+
+    with TestClient(app) as test_client:
+        yield test_client
 
 
 def valid_payload() -> dict:
@@ -23,7 +72,7 @@ def valid_payload() -> dict:
     }
 
 
-def test_health_endpoint() -> None:
+def test_health_endpoint(client: TestClient) -> None:
     response = client.get("/health")
 
     assert response.status_code == 200
@@ -32,7 +81,10 @@ def test_health_endpoint() -> None:
     }
 
 
-def test_valid_cdr_returns_202() -> None:
+def test_valid_cdr_returns_202(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
     payload = valid_payload()
 
     response = client.post(
@@ -47,8 +99,15 @@ def test_valid_cdr_returns_202() -> None:
         "raw_id": "raw-test-001",
     }
 
+    assert len(publisher.messages) == 1
 
-def test_missing_required_field_returns_400() -> None:
+    key, published_payload = publisher.messages[0]
+
+    assert key == "raw-test-001"
+    assert published_payload == payload
+
+
+def test_missing_required_field_returns_400(client: TestClient) -> None:
     payload = valid_payload()
     del payload["caller_num"]
 
@@ -65,7 +124,7 @@ def test_missing_required_field_returns_400() -> None:
     assert body["failed_field"] == "caller_num"
 
 
-def test_invalid_rat_returns_400() -> None:
+def test_invalid_rat_returns_400(client: TestClient) -> None:
     payload = valid_payload()
     payload["rat"] = 99
 
@@ -82,7 +141,7 @@ def test_invalid_rat_returns_400() -> None:
     assert body["failed_field"] == "rat"
 
 
-def test_invalid_epoch_returns_400() -> None:
+def test_invalid_epoch_returns_400(client: TestClient) -> None:
     payload = valid_payload()
 
     payload["epoch_start"] = 200
@@ -99,7 +158,8 @@ def test_invalid_epoch_returns_400() -> None:
 
     assert body["error"] == "invalid_epoch"
 
-def test_malformed_json_returns_400() -> None:
+
+def test_malformed_json_returns_400(client: TestClient) -> None:
     response = client.post(
         "/api/v1/cdr/raw",
         content='{"raw_id": invalid}',
@@ -114,7 +174,8 @@ def test_malformed_json_returns_400() -> None:
 
     assert body["error"] == "invalid_json"
 
-def test_non_object_json_returns_400() -> None:
+
+def test_non_object_json_returns_400(client: TestClient) -> None:
     response = client.post(
         "/api/v1/cdr/raw",
         json=[
@@ -129,3 +190,39 @@ def test_non_object_json_returns_400() -> None:
     body = response.json()
 
     assert body["error"] == "invalid_payload"
+
+
+def test_queue_full_returns_429(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
+    publisher.failure = IngestionQueueFull(
+        "Ingestion queue is full."
+    )
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=valid_payload(),
+    )
+
+    assert response.status_code == 429
+
+    assert response.json()["error"] == "backpressure"
+
+
+def test_kafka_failure_returns_503(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
+    publisher.failure = KafkaPublishError(
+        "Kafka unavailable."
+    )
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=valid_payload(),
+    )
+
+    assert response.status_code == 503
+
+    assert response.json()["error"] == "kafka_unavailable"