From 11bf3aa9281b5474faa05f2b78d622d6decc92a4 Mon Sep 17 00:00:00 2001 From: rustam810 Date: Wed, 29 Jul 2026 13:44:58 +0200 Subject: [PATCH] add ClickHouse DLQ support fix formatting change dlq data for all dlq producer --- design-docs/2026-08-dlq.md | 118 ++++++++++++++++++ packages/tkati-core/CHANGELOG.md | 6 + packages/tkati-core/tests/test_ch_producer.py | 61 +++++++-- .../tkati_core/clickhouse/producer.py | 36 +++++- packages/tkati-node-el/tests/test_node_el.py | 3 +- 5 files changed, 215 insertions(+), 9 deletions(-) create mode 100644 design-docs/2026-08-dlq.md diff --git a/design-docs/2026-08-dlq.md b/design-docs/2026-08-dlq.md new file mode 100644 index 0000000..1ebf5d0 --- /dev/null +++ b/design-docs/2026-08-dlq.md @@ -0,0 +1,118 @@ +# DLQ for `ClickhouseProducer` — retrospective design doc + +Status: retrospective — written after an initial implementation, to be revised and +used to align the code with the agreed design. + +## Problem + +`ClickhouseProducer.produce_arrow` inserts an Arrow batch into ClickHouse via +`clickhouse_connect`, retried up to 3 times (`_insert_with_retry`) to absorb +transient failures (network blips, brief unavailability). If all retries fail, +the exception previously propagated straight to the caller — a single +malformed or schema-incompatible row anywhere in the batch would fail the +*entire* batch, and by extension typically crash or stall the consuming node, +since a `Producer` failure has no other defined outcome. + +In a Kafka → ClickHouse pipeline, batches are read from a topic and can contain +many independent events. One event with a bad value (wrong type, oversized +string, invalid encoding, a `NULL` in a non-nullable column, etc.) shouldn't be +able to: + +1. Block delivery of every other, valid row in the same batch, or +2. Take down the node, which would stall consumption of all *subsequent* + batches too, since Kafka offsets can't advance past an unprocessed batch. + +We need a way to isolate and quarantine the specific row(s) causing an insert +to fail, while still delivering everything else, and without losing the +rejected data — it should be inspectable and potentially replayed later. + +## Non-goals + +* Retrying DLQ'd rows automatically. Once a row lands in the DLQ it's a human/ + offline concern. +* Deduplicating or rate-limiting DLQ writes. +* A DLQ for `KafkaProducer`/`KafkaConsumer` — this doc covers only + `ClickhouseProducer`, though the mechanism is intentionally generic (see + below) so it could extend to other producer types later. + +## Solution + +### Isolating bad rows: recursive bisection + +`ClickhouseProducer.produce_arrow` still attempts one `_insert_with_retry` for +the whole incoming batch. If that fails (all 3 attempts exhausted) and a +`dlq_producer` is configured, it falls back to `_insert_with_dlq_fallback`, +which bisects the batch recursively: + +* Batch of 1 row that fails → that row is the culprit. Wrap it as a DLQ + record (see schema below) and send it to `dlq_producer`. +* Batch of N>1 rows that fails → split it into chunks of + `max(1, N // split_factor)` rows and recurse on each chunk independently. + +This narrows down to exactly the failing rows without needing to inspect the +ClickHouse error message, at the cost of extra insert round-trips proportional +to how deep the recursion goes. `split_factor` (default 10, configurable via +`ClickHouseOutputSettings.dlq_split_factor`) trades off round-trip count +against how large a "wasted" chunk is when one bad row poisons a chunk insert: +a higher split factor means fewer rows re-sent per recursion level but more +levels of recursion for a large batch with a single bad row. + +If no `dlq_producer` is configured, a failed insert still raises after retries +are exhausted — this preserves today's fail-loud behavior for nodes that +haven't opted into a DLQ, rather than silently dropping data. + +### DLQ record schema + +Each rejected row is wrapped into a single-row Arrow table with a fixed +schema before being handed to `dlq_producer.produce_arrow`: + +| column | type | contents | +|---------------|--------|------------------------------------------------------------------| +| `producer` | string | JSON: `{"ch_url", "ch_database", "ch_table"}` — where it came from | +| `data` | string | JSON-encoded original row | +| `err_message` | string | `str(exception)` from the failed insert | +| `time` | datetime (UTC) | when the row was quarantined | + +This schema is independent of the source table's schema, so a single DLQ +sink (topic or table) can receive rejected rows from multiple source tables/ +producers — `producer` is what disambiguates provenance, and `data` carries +the arbitrary original row as JSON rather than requiring a matching column +layout. + +### `dlq_producer` is just a `Producer` + +The DLQ target is typed as `Producer` (`tkati_core.producer.Producer`), the +same interface used for regular node outputs — not a dedicated DLQ type. In +practice this means: + +* A `ClickhouseProducer` can be its own or another `ClickhouseProducer`'s + `dlq_producer` (writing quarantined rows to a `..._dlq` table), exercised by + `test_ch_producer_as_dlq_for_another_ch_producer`. +* A `KafkaProducer` could equally be used to push rejected rows to a DLQ + topic instead of a table, with no changes to `ClickhouseProducer`. + +`produce_arrow` calls `self._dlq_producer.flush()` once after the recursive +fallback completes (not once per DLQ'd row), so buffering producers (e.g. a +future Kafka DLQ) don't need a flush per row. + +### Wiring + +* `ClickHouseOutputSettings.dlq_split_factor: int = 10` — the only new setting + surfaced to config; the DLQ *target* itself is wired by the caller passing a + `dlq_producer` to `build_producer`/`ClickhouseProducer.__init__`, not by + settings. There's currently no config-driven way to say "this node's output + gets a DLQ pointed at table X" — that's assembled in application code. +* `build_producer` no longer takes a `split_factor` kwarg directly; it derives + it from `settings.dlq_split_factor` when settings is a + `ClickHouseOutputSettings`. + +## Open questions for the revised design + +* Should DLQ wiring become config-driven (e.g. an `[output.dlq]` settings + block resolved by `build_producer`) rather than assembled by hand in + application code? +* Is unbounded recursive bisection acceptable, or should there be a cap + (e.g. give up and DLQ the whole remaining chunk after N levels) to bound + worst-case round-trips against ClickHouse when many rows in a batch are bad? +* `KafkaProducer` has no DLQ fallback path today — is that in scope for this + effort or a later one? diff --git a/packages/tkati-core/CHANGELOG.md b/packages/tkati-core/CHANGELOG.md index e1b6b76..eccdb2e 100644 --- a/packages/tkati-core/CHANGELOG.md +++ b/packages/tkati-core/CHANGELOG.md @@ -1,3 +1,9 @@ +# WIP 0.8.4 + +* DLQ sinks now receive rejected rows using the DLQ schema + `producer`/`data`/`err_message`/`time`; `producer` contains the source + ClickHouse URL, database, and table as JSON + # 0.8.3 * `[input.config]` and `[output.config]` pass arbitrary librdkafka properties diff --git a/packages/tkati-core/tests/test_ch_producer.py b/packages/tkati-core/tests/test_ch_producer.py index 384137b..603201f 100644 --- a/packages/tkati-core/tests/test_ch_producer.py +++ b/packages/tkati-core/tests/test_ch_producer.py @@ -1,3 +1,4 @@ +import json from unittest.mock import MagicMock, patch import clickhouse_connect.driver.exceptions as ch_exc @@ -72,6 +73,30 @@ def test_a_data_error_is_not_retried() -> None: sleep.assert_not_called() +def _set_ch_client_metadata(ch_client: MagicMock) -> None: + ch_client.uri = "http://clickhouse:8123" + ch_client.database = "default" + + +def _assert_dlq_row( + sent: pa.Table, + *, + ch_table: str = "traffic_event", + err_message: str, + data: dict, +) -> None: + assert len(sent) == 1 + assert sent.column_names == ["producer", "data", "err_message", "time"] + row = sent.to_pylist()[0] + producer_info = json.loads(row["producer"]) + assert producer_info["ch_table"] == ch_table + assert producer_info["ch_database"] == "default" + assert producer_info["ch_url"] == "http://clickhouse:8123" + assert json.loads(row["data"]) == data + assert row["err_message"] == err_message + assert row["time"] is not None + + def test_insert_retry_on_non_data_error() -> None: """CH unreachable twice then back: insert_arrow called 3x.""" ch_client = MagicMock() @@ -131,6 +156,7 @@ def insert_side_effect(table, arrow_table): def test_dlq_single_bad_row() -> None: """A single row CH always rejects → written to DLQ once, without retries.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) dlq_producer = MagicMock() bad_row = _make_arrow_table(1) @@ -149,7 +175,7 @@ def test_dlq_single_bad_row() -> None: sleep.assert_not_called() dlq_producer.produce_arrow.assert_called_once() sent = dlq_producer.produce_arrow.call_args[0][0] - assert len(sent) == 1 + _assert_dlq_row(sent, err_message=str(_ch_data_error()), data={"uid": "uid-0", "traffic_in": 100}) def test_data_error_without_dlq_raises() -> None: @@ -167,6 +193,7 @@ def test_recursive_descent() -> None: """4-row batch rejected; with split_factor=2, recursion finds and DLQs exactly the one bad row, and the three good rows still reach ClickHouse.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) dlq_producer = MagicMock() def insert_side_effect(table, arrow_table): @@ -188,8 +215,11 @@ def insert_side_effect(table, arrow_table): dlq_producer.produce_arrow.assert_called_once() sent = dlq_producer.produce_arrow.call_args[0][0] - assert len(sent) == 1 - assert sent.column("uid")[0].as_py() == "uid-2" + _assert_dlq_row( + sent, + err_message=str(_ch_data_error()), + data={"uid": "uid-2", "traffic_in": 102}, + ) inserted = set() for call in ch_client.insert_arrow.call_args_list: @@ -203,6 +233,7 @@ def test_data_error_splits_without_sleeping() -> None: """Regression test for audit F3: isolating a bad row used to cost 2s of sleep at every level of the descent. It must now cost round-trips only.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) dlq_producer = MagicMock() def insert_side_effect(table, arrow_table): @@ -220,7 +251,11 @@ def insert_side_effect(table, arrow_table): sleep.assert_not_called() dlq_producer.produce_arrow.assert_called_once() - assert dlq_producer.produce_arrow.call_args[0][0].column("uid")[0].as_py() == "uid-7" + _assert_dlq_row( + dlq_producer.produce_arrow.call_args[0][0], + err_message=str(_ch_data_error()), + data={"uid": "uid-7", "traffic_in": 107}, + ) dlq_producer.flush.assert_called_once() @@ -271,6 +306,7 @@ def test_non_data_error_mid_descent_aborts_the_whole_descent() -> None: the DLQ stay sent, and are sent again once the rewound batch is re-read — the DLQ is at-least-once, like the output.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) dlq_producer = MagicMock() def insert_side_effect(table, arrow_table): @@ -292,7 +328,11 @@ def insert_side_effect(table, arrow_table): # uid-0 was isolated and filed before CH went away; uid-2 and uid-3 were # never reached as singles. assert dlq_producer.produce_arrow.call_count == 1 - assert dlq_producer.produce_arrow.call_args[0][0].column("uid")[0].as_py() == "uid-0" + _assert_dlq_row( + dlq_producer.produce_arrow.call_args[0][0], + err_message=str(_ch_data_error()), + data={"uid": "uid-0", "traffic_in": 100}, + ) dlq_producer.flush.assert_not_called() # uid-1 still gets its 3 attempts at that node: a blip mid-descent may pass, @@ -322,6 +362,7 @@ def test_ch_producer_success_no_dlq_call() -> None: def test_ch_producer_failure_with_dlq() -> None: """produce_arrow hits a bad row: recursive fallback runs and DLQ is flushed.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) dlq_producer = MagicMock() arrow_table = _make_arrow_table(1) ch_client.insert_arrow.side_effect = _ch_data_error() @@ -347,21 +388,26 @@ def test_ch_producer_failure_without_dlq_raises() -> None: def test_ch_producer_as_dlq_for_another_ch_producer() -> None: - """A ClickhouseProducer can itself be used as the dlq_producer for another ClickhouseProducer.""" + """A ClickhouseProducer used as DLQ receives a ClickHouse DLQ row.""" primary_ch_client = MagicMock() dlq_ch_client = MagicMock() arrow_table = _make_arrow_table(1) + _set_ch_client_metadata(primary_ch_client) primary_ch_client.insert_arrow.side_effect = _ch_data_error() dlq_producer = ClickhouseProducer(ch_client=dlq_ch_client, table="traffic_event_dlq") producer = ClickhouseProducer( - ch_client=primary_ch_client, table="traffic_event", dlq_producer=dlq_producer + ch_client=primary_ch_client, + table="traffic_event", + dlq_producer=dlq_producer, ) producer.produce_arrow(arrow_table) dlq_ch_client.insert_arrow.assert_called_once() + sent = dlq_ch_client.insert_arrow.call_args.kwargs["arrow_table"] + _assert_dlq_row(sent, err_message=str(_ch_data_error()), data={"uid": "uid-0", "traffic_in": 100}) def test_ch_producer_flush_is_noop() -> None: @@ -418,6 +464,7 @@ def test_ch_producer_does_not_pass_stats_to_its_dlq() -> None: """The DLQ fallback runs inside the primary's deliver block. Handing the DLQ the same stats would count that time twice.""" ch_client = MagicMock() + _set_ch_client_metadata(ch_client) ch_client.insert_arrow.side_effect = _ch_data_error() dlq_producer = MagicMock() stats = LoopStats(phases=PRODUCER_PHASES) diff --git a/packages/tkati-core/tkati_core/clickhouse/producer.py b/packages/tkati-core/tkati_core/clickhouse/producer.py index a5b7487..9090a34 100644 --- a/packages/tkati-core/tkati_core/clickhouse/producer.py +++ b/packages/tkati-core/tkati_core/clickhouse/producer.py @@ -1,6 +1,9 @@ +from datetime import UTC, datetime + import clickhouse_connect as ch import clickhouse_connect.driver as ch_driver import clickhouse_connect.driver.exceptions as ch_exc +import orjson import pyarrow as pa from loguru import logger from tenacity import ( @@ -114,6 +117,30 @@ def _table_slices(table: pa.Table, chunk_size: int) -> list[pa.Table]: return [table.slice(i, chunk_size) for i in range(0, len(table), chunk_size)] +def _make_dlq_table( + ch_client: ch_driver.Client, + ch_table: str, + row_table: pa.Table, + err: Exception, +) -> pa.Table: + return pa.Table.from_pylist( + [ + { + "producer": orjson.dumps( + { + "ch_url": ch_client.uri, + "ch_database": ch_client.database, + "ch_table": ch_table, + } + ).decode(), + "data": orjson.dumps(row_table.to_pylist()[0]).decode(), + "err_message": str(err), + "time": datetime.now(UTC), + } + ] + ) + + def _insert_with_dlq_fallback( table: pa.Table, ch_client: ch_driver.Client, @@ -139,7 +166,14 @@ def _insert_with_dlq_fallback( logger.error( f"Single row rejected by ClickHouse ({_describe(err)}), sending to DLQ" ) - dlq_producer.produce_arrow(table) + dlq_producer.produce_arrow( + _make_dlq_table( + ch_client, + ch_table, + table, + err, + ) + ) return chunk_size = max(1, len(table) // split_factor) logger.warning( diff --git a/packages/tkati-node-el/tests/test_node_el.py b/packages/tkati-node-el/tests/test_node_el.py index eb78cd0..a1ba819 100644 --- a/packages/tkati-node-el/tests/test_node_el.py +++ b/packages/tkati-node-el/tests/test_node_el.py @@ -138,7 +138,8 @@ def test_ch_producer_isolates_a_bad_row_against_real_clickhouse( mock_dlq_producer.produce_arrow.assert_called_once() sent = mock_dlq_producer.produce_arrow.call_args[0][0] - assert sent.column("uid").to_pylist() == ["not-a-uuid"] + assert sent.column_names == ["producer", "data", "err_message", "time"] + assert orjson.loads(sent.column("data")[0].as_py())["uid"] == "not-a-uuid" def test_every_batch_is_sent_unchanged() -> None: