diff --git a/.env.sample b/.env.sample index 462a8be..1a01ed2 100644 --- a/.env.sample +++ b/.env.sample @@ -72,6 +72,16 @@ CLICKHOUSE_USER=telecom CLICKHOUSE_PASSWORD=telecom_dev_password +# ========================= +# ClickHouse Reliability +# ========================= + +CLICKHOUSE_MAX_INSERT_ATTEMPTS=3 +CLICKHOUSE_RETRY_BASE_DELAY_SECONDS=0.5 + +CLICKHOUSE_CONNECT_TIMEOUT_SECONDS=5 +CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS=10 + # ========================= # Worker batching # ========================= diff --git a/README.md b/README.md index c6703e7..2b28edd 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,11 @@ worker processing is not connected yet. * Flush on 1,000 records or 2-second timeout * Kafka offsets committed only after successful ClickHouse batch writes * Remaining worker buffer flushed during graceful shutdown +* ClickHouse insert retries with exponential backoff +* Failed ClickHouse batches routed to the DLQ after three attempts +* Whole-batch offset commits with no partial Kafka acknowledgement +* Replay-safe storage using deterministic UUIDv5 identifiers and `ReplacingMergeTree` +* Graceful worker shutdown flushes buffered records before exit ## Project Structure @@ -205,3 +210,11 @@ pytest -v ``` The current test suite covers domain validation, normalization, deterministic ID generation, and API request validation. + +## Delivery Semantics + +The worker uses at-least-once delivery. + +Kafka offsets are committed only after a ClickHouse batch is successfully written or, after exhausted retries, the complete failed batch is acknowledged by the DLQ. + +Repeated records use deterministic UUIDv5 identifiers and `ReplacingMergeTree` to support replay-safe storage. diff --git a/src/core/config.py b/src/core/config.py index 39582ea..2454c6a 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -37,6 +37,12 @@ class Settings(BaseSettings): clickhouse_user: str = "telecom" clickhouse_password: str = "" + clickhouse_max_insert_attempts: int = 3 + clickhouse_retry_base_delay_seconds: float = 0.5 + + clickhouse_connect_timeout_seconds: int = 5 + clickhouse_send_receive_timeout_seconds: int = 10 + batch_size: int = 1_000 batch_flush_interval_seconds: float = 2.0 diff --git a/src/core/worker/batch.py b/src/core/worker/batch.py index 694efa1..98d6d35 100644 --- a/src/core/worker/batch.py +++ b/src/core/worker/batch.py @@ -1,5 +1,6 @@ import time from dataclasses import dataclass +from typing import Any from core.domain.models import NormalizedCDR @@ -14,6 +15,7 @@ class KafkaPosition: @dataclass(frozen=True, slots=True) class BufferedCDR: cdr: NormalizedCDR + original_payload: dict[str, Any] position: KafkaPosition diff --git a/src/core/worker/clickhouse.py b/src/core/worker/clickhouse.py index 96a753a..afb77ec 100644 --- a/src/core/worker/clickhouse.py +++ b/src/core/worker/clickhouse.py @@ -1,4 +1,6 @@ from collections.abc import Sequence +import logging +import time import clickhouse_connect @@ -6,6 +8,9 @@ from core.config import Settings from core.domain.models import NormalizedCDR +logger = logging.getLogger(__name__) + + class ClickHouseWriteError(Exception): pass @@ -40,6 +45,13 @@ class ClickHouseWriter: username=settings.clickhouse_user, password=settings.clickhouse_password, database=settings.clickhouse_database, + + connect_timeout=( + settings.clickhouse_connect_timeout_seconds + ), + send_receive_timeout=( + settings.clickhouse_send_receive_timeout_seconds + ), ) @staticmethod @@ -68,6 +80,22 @@ class ClickHouseWriter: str(cdr.source_ip), ] + def _insert_once( + self, + records: Sequence[NormalizedCDR], + ) -> None: + rows = [ + self._to_row(record) + for record in records + ] + + self._client.insert( + table="cdr_records", + data=rows, + column_names=self.COLUMN_NAMES, + database=self._settings.clickhouse_database, + ) + def insert( self, records: Sequence[NormalizedCDR], @@ -75,23 +103,65 @@ class ClickHouseWriter: if not records: return - rows = [ - self._to_row(record) - for record in records - ] + last_error: Exception | None = None - try: - self._client.insert( - table="cdr_records", - data=rows, - column_names=self.COLUMN_NAMES, - database=self._settings.clickhouse_database, + for attempt in range( + 1, + self._settings.clickhouse_max_insert_attempts + 1, + ): + try: + self._insert_once(records) + + if attempt > 1: + logger.info( + ( + "ClickHouse insert recovered. " + "attempt=%s batch_size=%s" + ), + attempt, + len(records), + ) + + return + + except Exception as exc: + last_error = exc + + logger.warning( + ( + "ClickHouse insert failed. " + "attempt=%s/%s batch_size=%s error=%s" + ), + attempt, + self._settings.clickhouse_max_insert_attempts, + len(records), + exc, + ) + + if ( + attempt + >= self._settings.clickhouse_max_insert_attempts + ): + break + + delay = ( + self._settings.clickhouse_retry_base_delay_seconds + * (2 ** (attempt - 1)) + ) + + logger.info( + "Retrying ClickHouse insert in %.2f seconds.", + delay, + ) + + time.sleep(delay) + + raise ClickHouseWriteError( + ( + "ClickHouse batch insert failed after " + f"{self._settings.clickhouse_max_insert_attempts} attempts." ) - - except Exception as exc: - raise ClickHouseWriteError( - f"ClickHouse batch insert failed: {exc}" - ) from exc + ) from last_error def close(self) -> None: self._client.close() diff --git a/src/core/worker/consumer.py b/src/core/worker/consumer.py index 4058dfc..1c9f8d6 100644 --- a/src/core/worker/consumer.py +++ b/src/core/worker/consumer.py @@ -13,7 +13,10 @@ from core.worker.batch import ( KafkaPosition, MicroBatch, ) -from core.worker.clickhouse import ClickHouseWriter +from core.worker.clickhouse import ( + ClickHouseWriteError, + ClickHouseWriter, +) from core.config import Settings from core.domain.errors import ( CDRProcessingError, @@ -171,6 +174,7 @@ class ProcessingWorker: self._batch.add( BufferedCDR( cdr=normalized, + original_payload=payload, position=position, ) ) @@ -257,19 +261,47 @@ class ProcessingWorker: for item in items ] + positions = [ + item.position + for item in items + ] + logger.info( "Flushing ClickHouse batch. size=%s", len(records), ) - self._clickhouse.insert( - records - ) + try: + self._clickhouse.insert( + records + ) - positions = [ - item.position - for item in items - ] + except ClickHouseWriteError as exc: + logger.exception( + "ClickHouse batch permanently failed." + ) + + self._route_batch_to_dlq( + items=items, + storage_error=exc, + ) + + # Only reached if EVERY DLQ publish succeeded. + self._commit_positions( + positions + ) + + self._batch.clear() + + logger.warning( + ( + "Failed ClickHouse batch routed to DLQ " + "and Kafka offsets committed. size=%s" + ), + len(records), + ) + + return self._commit_positions( positions @@ -332,6 +364,33 @@ class ProcessingWorker: partition.error ) + def _route_batch_to_dlq( + self, + items: list[BufferedCDR], + storage_error: ClickHouseWriteError, + ) -> None: + error = CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=( + "ClickHouse batch insert failed after retries: " + f"{storage_error}" + ), + ) + + logger.error( + ( + "Routing failed ClickHouse batch to DLQ. " + "batch_size=%s" + ), + len(items), + ) + + for item in items: + self._dlq.publish( + original_payload=item.original_payload, + error=error, + ) + def stop(self) -> None: self._running = False diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py index 53017d6..e5e7e4b 100644 --- a/tests/unit/test_batch.py +++ b/tests/unit/test_batch.py @@ -48,6 +48,11 @@ def buffered_item( ) -> BufferedCDR: return BufferedCDR( cdr=normalized_cdr(raw_id), + + original_payload={ + "raw_id": raw_id, + }, + position=KafkaPosition( topic="cdr.raw", partition=0,