Add k2ch to tkati

.github/workflow-templates/test.template.yml

@@ -63,7 +63,12 @@ jobs:
         run: uv sync --package {{ package.name }}
 
       - name: Check formatting
-        run: uv run ruff check {{ package.path }}
+        run: uvx ruff check {{ package.path }}
+
+      - name: Check typing
+        shell: bash
+        run: |
+          uv run mypy -p {{ package.package_name }}
 
       - name: Run tests
         run: uv run pytest {{ package.path }}

.github/workflows/test-k2ch.yml

@@ -0,0 +1,76 @@
+# This file was automatically generated by uv-workspace-codegen
+# For more information, see: https://github.com/epoch8/uv-workspace-codegen/blob/master/README.md
+# Do not edit this file manually - changes will be overwritten
+
+name: Test k2ch
+
+on:
+  push:
+    paths:
+      - "packages/tkati-k2ch/**"
+      - "packages/tkati-core/**"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+
+      - name: Build wheel
+        run: uv build
+        working-directory: packages/tkati-k2ch
+
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      redpanda:
+        image: redpandadata/redpanda:v25.2.1
+        ports:
+          - 9092:9092
+          - 9644:9644
+        options: >-
+          --health-cmd "curl -f http://localhost:9644/v1/status/ready"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+      # clickhouse:
+      #   image: clickhouse/clickhouse-server:latest
+      #   ports:
+      #     - 8123:8123
+      #     - 9000:9000
+      #   env:
+      #     CLICKHOUSE_DB: default
+      #     CLICKHOUSE_USER: default
+      #     CLICKHOUSE_PASSWORD: default
+      #   options: >-
+      #     --health-cmd "curl -f http://localhost:8123/ping"
+      #     --health-interval 5s
+      #     --health-timeout 5s
+      #     --health-retries 10
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+
+      - name: Install the latest version of uv
+        uses: astral-sh/setup-uv@v6
+
+      - name: Install deps
+        run: uv sync --package k2ch
+
+      - name: Check formatting
+        run: uvx ruff check packages/tkati-k2ch
+
+      - name: Check typing
+        shell: bash
+        run: |
+          uv run mypy -p k2ch
+
+      - name: Run tests
+        run: uv run pytest packages/tkati-k2ch

.github/workflows/test-tkati-core.yml

@@ -64,7 +64,12 @@ jobs:
         run: uv sync --package tkati-core
 
       - name: Check formatting
-        run: uv run ruff check packages/tkati-core
+        run: uvx ruff check packages/tkati-core
+
+      - name: Check typing
+        shell: bash
+        run: |
+          uv run mypy -p tkati_core
 
       - name: Run tests
         run: uv run pytest packages/tkati-core

packages/tkati-k2ch/README.md

@@ -0,0 +1,52 @@
+# k2ch — Kafka to ClickHouse
+
+Reads batches of JSON messages from a Kafka topic and inserts them into a ClickHouse table using the native Arrow protocol. Offsets are committed only after a successful insert (at-least-once delivery).
+
+## Configuration
+
+Settings are loaded from a TOML file. Set the `SETTINGS_FILE` environment variable to point to it (defaults to `settings.toml`).
+
+```toml
+[input.topic]
+broker = "redpanda:29092"
+name   = "traffic_event"
+
+[input.topic.schema]
+uid        = "string"
+time       = "timestamp[ms]"
+traffic_in = "uint32"
+# … other columns
+
+[input.consumer]
+group_id          = "k2ch-group"
+batch_size        = 1000
+batch_timeout_sec = 10
+auto_offset_reset = "latest"
+
+[output]
+host     = "clickhouse"
+port     = 9000
+user     = "default"
+password = ""
+database = "default"
+table    = "traffic_event"
+secure   = false
+
+[dlq]
+topic        = "k2ch-dlq"
+# broker defaults to input.topic.broker when omitted
+split_factor = 10
+```
+
+## DLQ semantics
+
+When a batch insert fails after all retries, the app switches to a recursive fallback to isolate the problematic rows:
+
+1. The failing batch is split into `split_factor` equal sub-batches and each is retried independently.
+2. If a sub-batch also fails it is split again — this repeats until individual rows are reached.
+3. A single row that ClickHouse still rejects is written to the DLQ Kafka topic in Arrow IPC (`arrow-batch`) format, preserving the full schema.
+4. After all rows are handled (inserted or DLQ'd), the Kafka offset is committed and the app resumes normal large-batch processing.
+
+With `split_factor=10` and a 1 000-row batch this takes at most 3 recursive levels (1000 → 100 → 10 → 1).
+
+**Delivery guarantee: at-least-once.** If the process crashes mid-recursion the uncommitted batch is re-read on restart and re-processed from the beginning, which may produce duplicate rows in ClickHouse and duplicate messages in the DLQ topic.

packages/tkati-k2ch/pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "k2ch"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "tkati-core",
+    "loguru>=0.7.0",
+    "pydantic-settings>=2.11.0",
+]
+
+[project.scripts]
+k2ch = "k2ch.main:main"
+
+[dependency-groups]
+dev = ["pytest>=9.0.1", "confluent-kafka>=2.11.0"]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+include = ["k2ch*"]
+
+[tool.uv]
+package = true
+
+[tool.uv-workspace-codegen]
+generate = true
+template_type = "test"

packages/tkati-k2ch/settings.test.toml

@@ -0,0 +1,33 @@
+[input.topic]
+broker = "redpanda:29092"
+name = "traffic_event"
+
+[input.topic.schema]
+uid = "string"
+time = "timestamp[ms]"
+package_id = "int32"
+user_hash = "string"
+sdk_hash = "string"
+conn_type = "string"
+country = "string"
+local_ip = "string"
+frontend_ip = "string"
+dest_addr = "string"
+client_ip = "string"
+traffic_in = "uint32"
+traffic_out = "uint32"
+
+[input.consumer]
+group_id = "k2ch-group"
+auto_offset_reset = "earliest"
+batch_size = 1000
+batch_timeout_sec = 10
+
+[output]
+host = "clickhouse"
+port = 8123
+user = "default"
+password = "default"
+database = "default"
+table = "traffic_event"
+secure = false

packages/tkati-k2ch/src/k2ch/__main__.py

@@ -0,0 +1,4 @@
+from .main import main
+
+if __name__ == "__main__":
+    main()

packages/tkati-k2ch/src/k2ch/main.py

@@ -0,0 +1,46 @@
+from loguru import logger
+from tkati_core.kafka.consumer import KafkaConsumer
+from tkati_core.kafka.producer import KafkaProducer
+
+from tkati_core.clickhouse.producer import ClickhouseProducer
+from k2ch.settings import AppSettings
+
+
+def run_one_iteration(
+    kafka_consumer: KafkaConsumer,
+    ch_producer: ClickhouseProducer,
+    settings: AppSettings,
+) -> None:
+    batch = kafka_consumer.read_arrow(
+        max_events_to_aggregate=settings.input.consumer.batch_size,
+        aggregation_interval_seconds=settings.input.consumer.batch_timeout_sec,
+    )
+    if batch is None:
+        return
+    ch_producer.produce_arrow(batch)
+    kafka_consumer.commit()
+    logger.info(f"Inserted {len(batch)} rows to {ch_producer._table}")
+
+
+def main() -> None:
+    settings = AppSettings()  # type: ignore
+
+    kafka_consumer = KafkaConsumer.from_input_settings(settings.input)
+
+    dlq_producer: KafkaProducer | None = None
+    if settings.dlq is not None:
+        dlq_producer = KafkaProducer.from_topic_settings(settings.dlq.topic)
+
+    ch_producer = ClickhouseProducer.from_output_settings(
+        settings=settings.output,
+        dlq_producer=dlq_producer,
+        split_factor=settings.dlq.split_factor if settings.dlq else 10,
+    )
+
+    try:
+        while True:
+            run_one_iteration(kafka_consumer, ch_producer, settings)
+    finally:
+        kafka_consumer.close()
+        if dlq_producer is not None:
+            dlq_producer.close()

packages/tkati-k2ch/src/k2ch/settings.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+from tkati_core.clickhouse.settings import ClickHouseOutputSettings
+from tkati_core.kafka.settings import KafkaInputSettings, KafkaTopicSettings
+from tkati_core.settings import TomlBaseSettings
+
+
+class DLQSettings(BaseModel):
+    topic: KafkaTopicSettings
+    split_factor: int = 10
+
+
+class AppSettings(TomlBaseSettings):
+    input: KafkaInputSettings
+    output: ClickHouseOutputSettings
+    dlq: DLQSettings | None = None

packages/tkati-k2ch/tests/conftest.py

@@ -0,0 +1,96 @@
+import uuid
+from collections.abc import Generator
+from unittest.mock import MagicMock
+
+import pytest
+from confluent_kafka import Producer
+from confluent_kafka.admin import AdminClient, NewTopic
+from k2ch.settings import AppSettings, ClickHouseOutputSettings, DLQSettings
+from tkati_core.kafka.settings import (
+    KafkaConsumerSettings,
+    KafkaInputSettings,
+    KafkaTopicSettings,
+)
+from tkati_core.kafka.testing import kafka_admin_client  # noqa: F401
+
+
+@pytest.fixture(scope="function")
+def test_settings() -> AppSettings:
+    run_id = str(uuid.uuid4())[:8]
+    return AppSettings(
+        input=KafkaInputSettings(
+            topic=KafkaTopicSettings(
+                broker="localhost:9092",
+                name=f"e2e_k2ch_{run_id}",
+                schema={
+                    "uid": "string",
+                    "time": "timestamp[ms]",
+                    "package_id": "int32",
+                    "user_hash": "string",
+                    "sdk_hash": "string",
+                    "conn_type": "string",
+                    "country": "string",
+                    "local_ip": "string",
+                    "frontend_ip": "string",
+                    "dest_addr": "string",
+                    "client_ip": "string",
+                    "traffic_in": "uint32",
+                    "traffic_out": "uint32",
+                },
+            ),
+            consumer=KafkaConsumerSettings(
+                group_id=f"test-k2ch-{run_id}",
+                auto_offset_reset="earliest",
+                batch_size=100,
+                batch_timeout_sec=5,
+            ),
+        ),
+        output=ClickHouseOutputSettings(
+            host="localhost",
+            port=8123,
+            user="default",
+            password="",
+            database="default",
+            table="traffic_event",
+            secure=False,
+        ),
+        dlq=DLQSettings(
+            topic=KafkaTopicSettings(broker="localhost:9092", name="k2ch-dlq"),
+            split_factor=2,
+        ),
+    )
+
+
+@pytest.fixture(scope="function")
+def mock_ch_client() -> MagicMock:
+    """Mock for clickhouse_connect client. insert_arrow is the only called method."""
+    client = MagicMock()
+    client.insert_arrow = MagicMock()
+    return client
+
+
+@pytest.fixture(scope="function")
+def mock_dlq_producer() -> MagicMock:
+    producer = MagicMock()
+    producer.produce_arrow = MagicMock()
+    producer.flush = MagicMock()
+    return producer
+
+
+@pytest.fixture(scope="function")
+def kafka_producer_and_topic(
+    test_settings: AppSettings,
+    kafka_admin_client: AdminClient,  # noqa: F811
+) -> Generator[Producer, None, None]:
+    """Creates the input topic and yields a Producer for it."""
+    topic = test_settings.input.topic.name
+    fs = kafka_admin_client.create_topics(
+        [NewTopic(topic, num_partitions=1, replication_factor=1)]
+    )
+    for t, f in fs.items():
+        try:
+            f.result()
+        except Exception as e:
+            print(f"Failed to create topic {t}: {e}")
+
+    yield Producer({"bootstrap.servers": test_settings.input.topic.broker})