diff --git a/scripts/functional_traffic.py b/scripts/functional_traffic.py
new file mode 100644
index 0000000..9a247ed
--- /dev/null
+++ b/scripts/functional_traffic.py
@@ -0,0 +1,242 @@
+import json
+import os
+import random
+import time
+import uuid
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from copy import deepcopy
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+
+BASE_URL = os.getenv(
+    "BASE_URL",
+    "http://localhost:8000",
+)
+
+ENDPOINT = f"{BASE_URL}/api/v1/cdr/raw"
+
+RUN_ID = os.getenv(
+    "RUN_ID",
+    uuid.uuid4().hex[:8],
+)
+
+
+def valid_payload(index: int) -> dict:
+    start = int(time.time())
+
+    return {
+        "raw_id": f"functional-{RUN_ID}-{index:04d}",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "+14155552671",
+        "svc": "4G_DATA",
+        "up_bytes": 1024,
+        "down_bytes": 4096,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": start,
+        "epoch_end": start + 300,
+    }
+
+
+def send_json(payload: dict) -> int | str:
+    body = json.dumps(
+        payload
+    ).encode("utf-8")
+
+    return send_raw(body)
+
+
+def send_raw(body: bytes) -> int | str:
+    request = Request(
+        ENDPOINT,
+        data=body,
+        method="POST",
+        headers={
+            "Content-Type": "application/json",
+        },
+    )
+
+    try:
+        with urlopen(
+            request,
+            timeout=10,
+        ) as response:
+            return response.status
+
+    except HTTPError as exc:
+        return exc.code
+
+    except URLError as exc:
+        return f"network_error:{exc.reason}"
+
+
+def main() -> None:
+    requests: list[
+        tuple[str, dict | bytes]
+    ] = []
+
+    valid_records = [
+        valid_payload(index)
+        for index in range(1000)
+    ]
+
+    # 1000 valid CDRs
+    for payload in valid_records:
+        requests.append(
+            ("valid", payload)
+        )
+
+    # 20 missing caller_num
+    for index in range(20):
+        payload = valid_payload(
+            2000 + index
+        )
+
+        payload.pop(
+            "caller_num"
+        )
+
+        requests.append(
+            ("missing_caller_num", payload)
+        )
+
+    # 10 malformed JSON requests
+    for _ in range(10):
+        requests.append(
+            (
+                "invalid_json",
+                b'{"raw_id": invalid}',
+            )
+        )
+
+    # 10 invalid RAT values
+    for index in range(10):
+        payload = valid_payload(
+            3000 + index
+        )
+
+        payload["rat"] = 99
+
+        requests.append(
+            ("invalid_rat", payload)
+        )
+
+    # 10 invalid epochs
+    for index in range(10):
+        payload = valid_payload(
+            4000 + index
+        )
+
+        payload["epoch_start"] = (
+            payload["epoch_end"] + 100
+        )
+
+        requests.append(
+            ("invalid_epoch", payload)
+        )
+
+    # 20 duplicates of already accepted valid records.
+    for payload in valid_records[:20]:
+        requests.append(
+            (
+                "duplicate",
+                deepcopy(payload),
+            )
+        )
+
+    random.shuffle(
+        requests
+    )
+
+    results = Counter()
+
+    started_at = time.perf_counter()
+
+    with ThreadPoolExecutor(
+        max_workers=50
+    ) as executor:
+        futures = []
+
+        for request_type, payload in requests:
+            if isinstance(
+                payload,
+                bytes,
+            ):
+                future = executor.submit(
+                    send_raw,
+                    payload,
+                )
+            else:
+                future = executor.submit(
+                    send_json,
+                    payload,
+                )
+
+            futures.append(
+                (
+                    request_type,
+                    future,
+                )
+            )
+
+        for request_type, future in futures:
+            status = future.result()
+
+            results[
+                (
+                    request_type,
+                    status,
+                )
+            ] += 1
+
+    elapsed = (
+        time.perf_counter()
+        - started_at
+    )
+
+    print()
+    print(
+        f"Run ID: {RUN_ID}"
+    )
+
+    print(
+        f"Total requests: {len(requests)}"
+    )
+
+    print(
+        f"Elapsed: {elapsed:.2f}s"
+    )
+
+    print()
+
+    for (
+        request_type,
+        status,
+    ), count in sorted(
+        results.items(),
+        key=lambda item: str(
+            item[0]
+        ),
+    ):
+        print(
+            f"{request_type:24} "
+            f"status={status!s:8} "
+            f"count={count}"
+        )
+
+    print()
+    print("Expected:")
+    print("  valid             → 1000 x 202")
+    print("  duplicate         →   20 x 202")
+    print("  missing_caller    →   20 x 400")
+    print("  invalid_json      →   10 x 400")
+    print("  invalid_rat       →   10 x 400")
+    print("  invalid_epoch     →   10 x 400")
+
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file
diff --git a/scripts/load_test.js b/scripts/load_test.js
new file mode 100644
index 0000000..673c1f5
--- /dev/null
+++ b/scripts/load_test.js
@@ -0,0 +1,131 @@
+import http from 'k6/http';
+import { check } from 'k6';
+import { Counter } from 'k6/metrics';
+
+
+const accepted = new Counter(
+    'cdr_accepted'
+);
+
+const backpressure = new Counter(
+    'cdr_backpressure'
+);
+
+const unavailable = new Counter(
+    'cdr_unavailable'
+);
+
+const unexpected = new Counter(
+    'cdr_unexpected'
+);
+
+
+const RATE = Number(
+    __ENV.RATE || 1000
+);
+
+const PRE_ALLOCATED_VUS = Number(
+    __ENV.PRE_ALLOCATED_VUS || 250
+);
+
+const MAX_VUS = Number(
+    __ENV.MAX_VUS || 1000
+);
+
+const DURATION = (
+    __ENV.DURATION || '10s'
+);
+
+const BASE_URL = (
+    __ENV.BASE_URL ||
+    'http://localhost:8000'
+);
+
+
+export const options = {
+    discardResponseBodies: true,
+
+    scenarios: {
+        ingestion: {
+            executor: 'constant-arrival-rate',
+
+            rate: RATE,
+            timeUnit: '1s',
+
+            duration: DURATION,
+
+            preAllocatedVUs: PRE_ALLOCATED_VUS,
+            maxVUs: MAX_VUS,
+        },
+    },
+};
+
+
+export default function () {
+    const now = Math.floor(
+        Date.now() / 1000
+    );
+
+    const rawId =
+        `k6-${__VU}-${__ITER}-${Date.now()}`;
+
+    const payload = JSON.stringify({
+        raw_id: rawId,
+
+        src_ip: '10.240.12.1',
+
+        device_id:
+            '310410998877665',
+
+        caller_num:
+            '+14155552671',
+
+        svc: '4G_DATA',
+
+        up_bytes: 1024,
+        down_bytes: 4096,
+
+        rat: 6,
+
+        lac: '4021',
+        ci: '8812',
+
+        epoch_start: now,
+        epoch_end: now + 300,
+    });
+
+    const response = http.post(
+        `${BASE_URL}/api/v1/cdr/raw`,
+        payload,
+        {
+            headers: {
+                'Content-Type':
+                    'application/json',
+            },
+
+            tags: {
+                endpoint: 'cdr_ingest',
+            },
+        }
+    );
+
+    if (response.status === 202) {
+        accepted.add(1);
+
+    } else if (response.status === 429) {
+        backpressure.add(1);
+
+    } else if (response.status === 503) {
+        unavailable.add(1);
+
+    } else {
+        unexpected.add(1);
+    }
+
+    check(response, {
+        'accepted or gracefully rejected':
+            (r) =>
+                r.status === 202 ||
+                r.status === 429,
+    });
+}
diff --git a/tree b/tree
new file mode 100644
index 0000000..3ecc035
--- /dev/null
+++ b/tree
@@ -0,0 +1,772 @@
+# Telecom CDR Pipeline
+
+High-throughput, fault-tolerant telecom Call Detail Record ingestion and analytics pipeline built with FastAPI, Redpanda/Kafka, ClickHouse, and Python.
+
+## Architecture
+
+```text
+Network Element
+      │
+      ▼
+FastAPI Ingestion API
+      │
+      ├── Validation
+      │
+      ├── Bounded Async Queue
+      │
+      └── Kafka Producer
+      │
+      ▼
+Redpanda / cdr.raw
+      │
+      ▼
+Processing Worker
+      │
+      ├── Decode + Validate
+      ├── Normalize
+      ├── Micro-Batch
+      │
+      ├── valid ──────► ClickHouse
+      │
+      └── invalid ────► cdr.raw.dlq
+```
+
+The main services are:
+
+* **FastAPI API** — validates incoming CDRs, applies backpressure, and publishes accepted records to Kafka.
+* **Redpanda** — Kafka-compatible durable buffer between ingestion and processing.
+* **Processing Worker** — validates, normalizes, batches, and persists CDRs.
+* **ClickHouse** — analytical storage for normalized records.
+* **DLQ** — stores malformed or unprocessable records with structured failure information.
+
+## Features
+
+* Bounded asynchronous ingestion queue
+* `202`, `400`, `429`, and `503` ingestion responses
+* Long-lived asynchronous Kafka producer
+* Kafka delivery acknowledgement before `202 Accepted`
+* Kafka consumer with manual offset management
+* Worker-side validation and normalization
+* E.164 phone-number normalization using `phonenumbers`
+* RAT mapping for `3G`, `LTE_4G`, and `5G_NR`
+* Deterministic UUIDv5 `cdr_id` generation
+* Structured Dead Letter Queue
+* Configurable ClickHouse micro-batching
+* Flush on 1,000 records or 2-second timeout
+* ClickHouse retries with exponential backoff
+* At-least-once delivery semantics
+* Replay-safe ClickHouse storage using `ReplacingMergeTree`
+* Graceful worker shutdown with final buffer flush
+* Separate Prometheus metrics endpoints for API and worker
+* Kafka consumer lag monitoring
+* Aggregated `/stats` endpoint
+* Health-based Docker Compose startup ordering
+* Automated Kafka topic provisioning
+* Automated and idempotent ClickHouse initialization
+* Unit, integration, functional, and load testing
+
+---
+
+## Project Structure
+
+Generated directories such as `__pycache__` and `*.egg-info` are omitted below.
+
+```text
+telecom-cdr-pipeline/
+├── Dockerfile.api
+├── Dockerfile.worker
+├── README.md
+├── docker-compose.yml
+├── pyproject.toml
+│
+├── infrastructure/
+│   └── clickhouse/
+│       └── init.sql
+│
+├── scripts/
+│   ├── functional_traffic.py
+│   ├── load_test.js
+│   └── sample-valid-cdr.json
+│
+├── src/
+│   └── core/
+│       ├── __init__.py
+│       ├── config.py
+│       │
+│       ├── api/
+│       │   ├── __init__.py
+│       │   ├── kafka_publisher.py
+│       │   ├── main.py
+│       │   ├── metrics.py
+│       │   ├── publisher.py
+│       │   ├── routes.py
+│       │   └── stats.py
+│       │
+│       ├── domain/
+│       │   ├── __init__.py
+│       │   ├── errors.py
+│       │   ├── models.py
+│       │   ├── normalization.py
+│       │   └── validation.py
+│       │
+│       └── worker/
+│           ├── __init__.py
+│           ├── batch.py
+│           ├── clickhouse.py
+│           ├── consumer.py
+│           ├── dlq.py
+│           ├── main.py
+│           ├── metrics.py
+│           ├── observability.py
+│           ├── processor.py
+│           └── stats.py
+│
+└── tests/
+    ├── integration/
+    │   └── test_ingestion_api.py
+    │
+    └── unit/
+        ├── test_batch.py
+        ├── test_dlq.py
+        ├── test_normalization.py
+        ├── test_project_setup.py
+        ├── test_validation.py
+        ├── test_worker_processor.py
+        └── test_worker_stats.py
+```
+
+---
+
+## Quickstart
+
+### 1. Configure Environment
+
+Create your local environment file from `.env.example`:
+
+```bash
+cp .env.example .env
+```
+
+### 2. Start the Complete Pipeline
+
+```bash
+docker compose up --build -d
+```
+
+Check service status:
+
+```bash
+docker compose ps -a
+```
+
+Expected state:
+
+```text
+telecom-redpanda        healthy
+telecom-clickhouse      healthy
+telecom-api             healthy
+telecom-worker          healthy
+telecom-redpanda-init   exited (0)
+```
+
+### 3. Verify Health
+
+API:
+
+```bash
+curl http://localhost:8000/health
+```
+
+Worker:
+
+```bash
+curl http://localhost:8001/health
+```
+
+### 4. Send a Sample CDR
+
+```bash
+curl -X POST \
+  http://localhost:8000/api/v1/cdr/raw \
+  -H "Content-Type: application/json" \
+  --data-binary "@scripts/sample-valid-cdr.json"
+```
+
+Expected response:
+
+```json
+{
+  "status": "accepted",
+  "raw_id": "raw-test-001"
+}
+```
+
+---
+
+## API Endpoints
+
+### Ingestion API
+
+```text
+POST /api/v1/cdr/raw
+```
+
+Possible responses:
+
+| Status | Meaning                                     |
+| ------ | ------------------------------------------- |
+| `202`  | Payload validated and acknowledged by Kafka |
+| `400`  | Invalid JSON or invalid CDR schema          |
+| `429`  | Bounded ingestion queue is full             |
+| `503`  | Kafka delivery is unavailable               |
+
+### API Observability
+
+```text
+GET http://localhost:8000/health
+GET http://localhost:8000/metrics
+GET http://localhost:8000/stats
+```
+
+### Worker Observability
+
+```text
+GET http://localhost:8001/health
+GET http://localhost:8001/metrics
+GET http://localhost:8001/internal/stats
+```
+
+---
+
+## Micro-Batching
+
+The worker never inserts CDRs into ClickHouse row-by-row.
+
+Normalized records are buffered and flushed when either condition occurs:
+
+```text
+Batch reaches 1,000 records
+            OR
+2 seconds pass after the batch starts
+```
+
+Configuration:
+
+```env
+BATCH_SIZE=1000
+BATCH_FLUSH_INTERVAL_SECONDS=2
+```
+
+This reduces ClickHouse insert overhead while still keeping low-volume traffic latency bounded.
+
+On `SIGTERM`, the worker stops consuming new records, flushes the remaining buffer, commits the corresponding Kafka offsets, and then exits.
+
+---
+
+## Delivery and Offset Strategy
+
+The worker uses **at-least-once delivery**.
+
+Kafka auto-commit is disabled.
+
+The normal processing order is:
+
+```text
+Kafka consume
+      ↓
+Validate + Normalize
+      ↓
+Micro-Batch
+      ↓
+ClickHouse INSERT
+      ↓
+ClickHouse ACK
+      ↓
+Kafka offset commit
+```
+
+Offsets are never committed before the corresponding processing outcome is safely completed.
+
+### ClickHouse Failure
+
+A failed ClickHouse batch is retried with exponential backoff:
+
+```text
+Attempt 1
+   ↓ failure
+0.5s delay
+
+Attempt 2
+   ↓ failure
+1.0s delay
+
+Attempt 3
+   ↓ failure
+Entire batch → DLQ
+```
+
+Offsets are committed only after every DLQ publication succeeds.
+
+If DLQ publishing fails, the offsets remain uncommitted so Kafka can replay the records.
+
+### Worker Crash After ClickHouse Write
+
+If ClickHouse successfully stores a batch but the worker crashes before committing Kafka offsets, Kafka delivers the records again after restart.
+
+This can produce duplicate physical inserts, which is handled by the deduplication strategy below.
+
+### Partial Batch Failure
+
+A batch is treated as a single processing unit.
+
+Offsets are never partially committed for a failed ClickHouse batch.
+
+---
+
+## Deduplication
+
+Each CDR gets a deterministic UUIDv5 generated from its `raw_id`:
+
+```python
+uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)
+```
+
+The same `raw_id` therefore always generates the same `cdr_id`.
+
+ClickHouse uses:
+
+```text
+ReplacingMergeTree(processed_at)
+```
+
+with `cdr_id` included in the sorting key.
+
+This allows records replayed because of at-least-once Kafka delivery to eventually collapse into one logical record.
+
+For immediate query-time deduplication:
+
+```sql
+SELECT *
+FROM telecom.cdr_records FINAL;
+```
+
+---
+
+## Dead Letter Queue
+
+Invalid or unprocessable Kafka records are published to:
+
+```text
+cdr.raw.dlq
+```
+
+DLQ records include:
+
+```json
+{
+  "original_payload": {},
+  "failure_reason": "invalid_rat_value",
+  "failed_field": "rat",
+  "failed_at": "2026-09-20T12:00:00Z",
+  "worker_instance": "worker-1"
+}
+```
+
+Supported failure reasons:
+
+```text
+missing_required_field
+invalid_rat_value
+invalid_epoch
+malformed_json
+normalization_error
+```
+
+A bad record does not crash the worker.
+
+---
+
+## Monitoring
+
+### API Metrics
+
+```bash
+curl http://localhost:8000/metrics
+```
+
+Includes:
+
+```text
+cdr_api_received_total
+```
+
+### Worker Metrics
+
+```bash
+curl http://localhost:8001/metrics
+```
+
+Includes:
+
+```text
+cdr_processed_total
+cdr_dlq_total
+
+cdr_kafka_to_clickhouse_latency_seconds
+cdr_batch_flush_duration_seconds
+cdr_batch_size_at_flush
+
+cdr_in_flight_buffer_size
+cdr_kafka_consumer_lag
+```
+
+### Pipeline Stats
+
+```bash
+curl http://localhost:8000/stats
+```
+
+Example:
+
+```json
+{
+  "pipeline": {
+    "total_received": 1000,
+    "total_processed": 995,
+    "total_dlq": 5,
+    "in_flight_buffer": 0,
+    "last_flush_at": "2026-09-20T12:00:00Z"
+  },
+  "breakdown": {
+    "by_service_type": {},
+    "by_radio_access_type": {},
+    "dlq_by_reason": {}
+  }
+}
+```
+
+---
+
+## ClickHouse Verification
+
+Open a ClickHouse client:
+
+```bash
+docker compose exec clickhouse \
+  clickhouse-client \
+  --user telecom \
+  --password telecom_dev_password \
+  --database telecom
+```
+
+Count records by service:
+
+```sql
+SELECT
+    service_type,
+    count() AS total
+FROM cdr_records
+GROUP BY service_type;
+```
+
+Average duration by RAT:
+
+```sql
+SELECT
+    radio_access_type,
+    avg(duration_seconds) AS avg_duration
+FROM cdr_records
+GROUP BY radio_access_type;
+```
+
+Top cells by traffic:
+
+```sql
+SELECT
+    cell_id,
+    sum(total_bytes) AS total_traffic
+FROM cdr_records
+GROUP BY cell_id
+ORDER BY total_traffic DESC
+LIMIT 10;
+```
+
+Check logical deduplication:
+
+```sql
+SELECT
+    raw_id,
+    count()
+FROM cdr_records FINAL
+GROUP BY raw_id
+HAVING count() > 1;
+```
+
+Expected result:
+
+```text
+0 rows
+```
+
+---
+
+## Automated Tests
+
+Install development dependencies:
+
+```bash
+pip install -e ".[dev]"
+```
+
+Run all unit and integration tests:
+
+```bash
+pytest -v
+```
+
+---
+
+## Functional Traffic Verification
+
+The functional traffic script exercises the pipeline with the assessment scenarios, including valid records, invalid requests, and duplicate `raw_id` values.
+
+Run:
+
+```bash
+python scripts/functional_traffic.py
+```
+
+The assessment scenarios are:
+
+| Scenario                  | Count | Expected                                      |
+| ------------------------- | ----: | --------------------------------------------- |
+| Valid records             | 1,000 | Stored in ClickHouse                          |
+| Missing `caller_num`      |    20 | Rejected / DLQ when worker is tested directly |
+| Invalid JSON              |    10 | API `400`; DLQ when worker is bypassed        |
+| Unknown `rat=99`          |    10 | Rejected / DLQ when worker is tested directly |
+| `epoch_start > epoch_end` |    10 | Rejected / DLQ when worker is tested directly |
+| Duplicate `raw_id`        |    20 | One logical ClickHouse record per `raw_id`    |
+
+Invalid records submitted through the public API are rejected before Kafka. Worker-side DLQ handling is independently exercised by publishing malformed or invalid records directly to `cdr.raw` and by the automated worker tests.
+
+### Functional Traffic Result
+
+Paste the final `functional_traffic.py` output below:
+
+```text
+Total requests: 1070
+Elapsed: 5.21s
+
+duplicate                status=202      count=20
+invalid_epoch            status=400      count=10
+invalid_json             status=400      count=10
+invalid_rat              status=400      count=10
+missing_caller_num       status=400      count=20
+valid                    status=202      count=1000
+
+Expected:
+  valid             → 1000 x 202
+  duplicate         →   20 x 202
+  missing_caller    →   20 x 400
+  invalid_json      →   10 x 400
+  invalid_rat       →   10 x 400
+  invalid_epoch     →   10 x 400
+```
+
+---
+
+## Required k6 Burst Test
+
+The assessment requires:
+
+```text
+500 records in less than 5 seconds
+0 × 429 responses
+No data loss
+```
+
+The included k6 script uses a configurable constant arrival rate.
+
+Run the required assessment scenario with:
+
+```bash
+docker run --rm -i \
+  --network=host \
+  -e RATE=125 \
+  -e DURATION=4s \
+  -e PRE_ALLOCATED_VUS=500 \
+  -e MAX_VUS=750 \
+  grafana/k6 run - \
+  < scripts/load_test.js
+```
+
+This schedules approximately:
+
+```text
+125 requests/second × 4 seconds = 500 requests
+```
+
+A successful result should show approximately:
+
+```text
+iterations             500
+http_reqs              500
+cdr_accepted           500
+cdr_backpressure       0
+cdr_unavailable        0
+http_req_failed        0%
+dropped_iterations     0
+```
+
+After the run, allow the worker to finish processing and verify that no records were lost.
+
+Check worker lag:
+
+```bash
+curl -s http://localhost:8001/metrics \
+  | grep cdr_kafka_consumer_lag
+```
+
+Check pipeline statistics:
+
+```bash
+curl http://localhost:8000/stats
+```
+
+### Required k6 Result
+
+Paste the final required 500-record burst result below:
+
+```text
+    TOTAL RESULTS 
+
+    checks_total.......: 4087   272.979956/s
+    checks_succeeded...: 34.47% 1409 out of 4087
+    checks_failed......: 65.52% 2678 out of 4087
+
+    ✗ accepted or gracefully rejected
+      ↳  34% — ✓ 1409 / ✗ 2678
+
+    CUSTOM
+    cdr_accepted...................: 1409   94.110291/s
+    cdr_unavailable................: 2678   178.869665/s
+
+    HTTP
+    http_req_duration..............: avg=5.26s min=44.55ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
+      { expected_response:true }...: avg=2.33s min=44.55ms med=1.75s max=9.94s p(90)=5.75s  p(95)=6.1s 
+    http_req_failed................: 65.52% 2678 out of 4087
+    http_reqs......................: 4087   272.979956/s
+
+    EXECUTION
+    dropped_iterations.............: 5912   394.875826/s
+    iteration_duration.............: avg=5.26s min=46.48ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
+    iterations.....................: 4087   272.979956/s
+    vus............................: 214    min=214          max=2000
+    vus_max........................: 2000   min=1500         max=2000
+
+    NETWORK
+    data_received..................: 862 kB 58 kB/s
+    data_sent......................: 1.6 MB 105 kB/s
+
+running (15.0s), 0000/2000 VUs, 4087 complete and 0 interrupted iterations
+ingestion ✓ [ 100% ] 0000/2000 VUs  10s  1000.00 iters/s
+```
+
+---
+
+## Optional Stress Test
+
+The required assessment test is the 500-record burst above.
+
+A larger stress test can optionally be executed with:
+
+```bash
+docker run --rm -i \
+  --network=host \
+  -e RATE=1000 \
+  -e DURATION=10s \
+  -e PRE_ALLOCATED_VUS=1500 \
+  -e MAX_VUS=2000 \
+  grafana/k6 run - \
+  < scripts/load_test.js
+```
+
+This is useful for identifying throughput and latency limits but is not required for the assessment's 500-record burst criterion.
+
+---
+
+## Configuration Reference
+
+Configuration is provided through environment variables. Local defaults are documented in `.env.example`.
+
+| Variable                                  | Default                                | Purpose                                |
+| ----------------------------------------- | -------------------------------------- | -------------------------------------- |
+| `REDPANDA_VERSION`                        | `v26.2.3`                              | Redpanda image version                 |
+| `KAFKA_BOOTSTRAP_SERVERS`                 | `localhost:19092`                      | Kafka address for host applications    |
+| `KAFKA_INTERNAL_BOOTSTRAP_SERVERS`        | `redpanda:9092`                        | Kafka address inside Docker            |
+| `KAFKA_RAW_TOPIC`                         | `cdr.raw`                              | Raw CDR topic                          |
+| `KAFKA_RAW_TOPIC_PARTITIONS`              | `3`                                    | Raw topic partitions                   |
+| `KAFKA_RAW_TOPIC_REPLICATION_FACTOR`      | `1`                                    | Raw topic replication                  |
+| `KAFKA_RAW_TOPIC_RETENTION_MS`            | `86400000`                             | Raw topic retention                    |
+| `KAFKA_RAW_TOPIC_COMPRESSION`             | `lz4`                                  | Raw topic compression                  |
+| `KAFKA_DLQ_TOPIC`                         | `cdr.raw.dlq`                          | Dead Letter Queue topic                |
+| `KAFKA_DLQ_TOPIC_PARTITIONS`              | `1`                                    | DLQ partitions                         |
+| `KAFKA_DLQ_TOPIC_REPLICATION_FACTOR`      | `1`                                    | DLQ replication                        |
+| `KAFKA_DLQ_TOPIC_RETENTION_MS`            | `604800000`                            | DLQ retention                          |
+| `KAFKA_CONSUMER_GROUP`                    | `cdr-processing-workers`               | Worker consumer group                  |
+| `KAFKA_AUTO_OFFSET_RESET`                 | `earliest`                             | Consumer fallback offset               |
+| `KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS`     | `1`                                    | Worker Kafka poll timeout              |
+| `KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS`      | `5`                                    | DLQ delivery timeout                   |
+| `KAFKA_DELIVERY_TIMEOUT_SECONDS`          | `4`                                    | API Kafka acknowledgement timeout      |
+| `KAFKA_MESSAGE_TIMEOUT_MS`                | `3000`                                 | Kafka message timeout                  |
+| `KAFKA_LINGER_MS`                         | `5`                                    | Producer batching delay                |
+| `KAFKA_BATCH_SIZE`                        | `131072`                               | Kafka producer batch size              |
+| `KAFKA_PRODUCER_MAX_MESSAGES`             | `100000`                               | Kafka producer internal queue limit    |
+| `API_QUEUE_SIZE`                          | `10000`                                | Bounded API ingestion queue            |
+| `CLICKHOUSE_VERSION`                      | `26.8.7.19`                            | ClickHouse image version               |
+| `CLICKHOUSE_HOST`                         | `localhost`                            | ClickHouse host for local applications |
+| `CLICKHOUSE_INTERNAL_HOST`                | `clickhouse`                           | ClickHouse Docker service hostname     |
+| `CLICKHOUSE_PORT`                         | `8123`                                 | ClickHouse HTTP port                   |
+| `CLICKHOUSE_HTTP_HOST_PORT`               | `8123`                                 | Host-exposed ClickHouse port           |
+| `CLICKHOUSE_DATABASE`                     | `telecom`                              | ClickHouse database                    |
+| `CLICKHOUSE_USER`                         | `telecom`                              | ClickHouse user                        |
+| `CLICKHOUSE_PASSWORD`                     | `telecom_dev_password`                 | Local development password             |
+| `BATCH_SIZE`                              | `1000`                                 | Worker micro-batch maximum size        |
+| `BATCH_FLUSH_INTERVAL_SECONDS`            | `2`                                    | Worker time-based flush interval       |
+| `CLICKHOUSE_MAX_INSERT_ATTEMPTS`          | `3`                                    | Maximum ClickHouse insert attempts     |
+| `CLICKHOUSE_RETRY_BASE_DELAY_SECONDS`     | `0.5`                                  | Exponential retry base delay           |
+| `CLICKHOUSE_CONNECT_TIMEOUT_SECONDS`      | `5`                                    | ClickHouse connection timeout          |
+| `CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS` | `10`                                   | ClickHouse request timeout             |
+| `WORKER_INSTANCE_ID`                      | `worker-1`                             | Worker instance identifier             |
+| `WORKER_HTTP_HOST`                        | `0.0.0.0`                              | Worker observability bind address      |
+| `WORKER_HTTP_PORT`                        | `8001`                                 | Worker observability port              |
+| `WORKER_STATS_URL`                        | `http://localhost:8001/internal/stats` | Worker stats URL for host-run API      |
+| `WORKER_STATS_TIMEOUT_SECONDS`            | `2`                                    | Worker stats HTTP timeout              |
+| `API_HOST_PORT`                           | `8000`                                 | API host port                          |
+| `WORKER_HTTP_HOST_PORT`                   | `8001`                                 | Worker observability host port         |
+
+Inside Docker Compose, service-to-service addresses override host defaults where required:
+
+```text
+Kafka:      redpanda:9092
+ClickHouse: clickhouse:8123
+Worker:     http://worker:8001
+```
+
+---
+
+## Shutdown
+
+Stop the stack while preserving Kafka and ClickHouse data:
+
+```bash
+docker compose down
+```
+
+Remove containers and persistent development data:
+
+```bash
+docker compose down -v
+```
+
+Use `-v` only when a completely clean local environment is desired.