validating-clickhouse-kafka-pipelines

What this skill does

# Validating ClickHouse + Kafka Pipelines

## Purpose

This skill provides production-ready patterns for implementing defense-in-depth validation in ClickHouse + Kafka data pipelines. It ensures data integrity at both producer and consumer boundaries, handles errors without blocking consumption, automatically deduplicates messages, and provides comprehensive monitoring and observability.

## When to Use This Skill

**Explicit Triggers:**
- "Validate kafka pipeline"
- "Implement clickhouse error handling"
- "Set up kafka deduplication"
- "Create anti-corruption layer for kafka"
- "Configure kafka error streaming"

**Implicit Triggers:**
- Building new data pipeline with Kafka + ClickHouse
- Experiencing duplicate messages in ClickHouse
- Kafka consumption blocked by malformed messages
- Need defense-in-depth validation strategy
- Implementing event-driven architecture with data integrity requirements

**Debugging Scenarios:**
- Messages disappearing from Kafka but not in ClickHouse
- Duplicate records despite idempotent producer
- ClickHouse consumer stuck on malformed JSON
- Type validation failures causing data loss
- Schema evolution breaking existing pipelines

## Quick Start

### 1. Enable Error Streaming in ClickHouse (30 seconds)

```sql
-- Add error streaming to your Kafka table
ALTER TABLE kafka_orders
MODIFY SETTING kafka_handle_error_mode = 'stream';

-- Create error capture table
CREATE TABLE IF NOT EXISTS kafka_errors (
    topic String,
    partition Int64,
    offset Int64,
    raw_message String,
    error_message String,
    captured_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
ORDER BY (topic, partition, offset)
PARTITION BY toYYYYMM(captured_at)
TTL captured_at + INTERVAL 30 DAY;

-- Capture errors via materialized view
CREATE MATERIALIZED VIEW IF NOT EXISTS kafka_errors_mv
TO kafka_errors AS
SELECT
    _topic AS topic,
    _partition AS partition,
    _offset AS offset,
    _raw_message AS raw_message,
    _error AS error_message
FROM kafka_orders
WHERE length(_error) > 0;
```

### 2. Set Up Idempotency with ReplacingMergeTree

```sql
-- Target table with automatic deduplication
CREATE TABLE IF NOT EXISTS orders (
    order_id String,
    created_at DateTime,
    line_items Array(Tuple(
        line_item_id String,
        product_id String,
        product_title String,
        quantity Int32
    )),
    inserted_at DateTime
)
ENGINE = ReplacingMergeTree()
ORDER BY (order_id, created_at)
PARTITION BY toYYYYMM(created_at);

-- Valid message transformation
CREATE MATERIALIZED VIEW IF NOT EXISTS orders_mv
TO orders AS
SELECT
    order_id,
    parseDateTime64BestEffort(created_at) AS created_at,
    line_items,
    parseDateTime64BestEffort(inserted_at) AS inserted_at
FROM kafka_orders
WHERE length(_error) = 0;
```

### 3. Validate at Producer (Python - msgspec)

```python
# app/extraction/adapters/kafka/schemas.py
from __future__ import annotations

import msgspec
from datetime import datetime, timezone


class LineItemMessage(msgspec.Struct, frozen=True):
    """Type-safe schema for line items."""
    line_item_id: str
    product_id: str
    product_title: str
    quantity: int


class OrderMessage(msgspec.Struct, frozen=True):
    """Type-safe schema for orders."""
    order_id: str
    created_at: str  # ISO format
    line_items: list[LineItemMessage]
    inserted_at: str  # ISO format


# Publish with automatic validation
from app.core.monitoring.otel_logger import logger, traced

logger = logger(__name__)


@traced
def publish_order(order: Order, producer: KafkaProducer) -> None:
    """Publish order with msgspec validation."""
    try:
        message = OrderMessage(
            order_id=str(order.order_id),
            created_at=order.created_at.isoformat(),
            line_items=[
                LineItemMessage(
                    line_item_id=item.line_item_id,
                    product_id=str(item.product_id),
                    product_title=str(item.product_title),
                    quantity=item.quantity
                )
                for item in order.line_items
            ],
            inserted_at=datetime.now(timezone.utc).isoformat()
        )

        payload = msgspec.json.encode(message)
        producer.produce(topic="shopify-orders", value=payload)
        logger.info("order_published", order_id=order.order_id)

    except msgspec.ValidationError as e:
        logger.error("order_validation_failed", error=str(e), order_id=order.order_id)
        raise InvalidOrderException(f"Order validation failed: {e}") from e
```

### 4. Validate at Consumer (Python - Anti-Corruption Layer)

```python
# app/storage/adapters/anti_corruption.py
from __future__ import annotations

import msgspec
from datetime import datetime
from app.core.monitoring.otel_logger import logger, traced
from app.storage.adapters.kafka.schemas import OrderMessage
from app.storage.domain.entities import Order, OrderItem

logger = logger(__name__)


class KafkaMessageTranslator:
    """Anti-corruption layer with defense-in-depth validation."""


    @staticmethod
    @traced
    def translate_order(message: OrderMessage) -> Order:
        """Translate and validate Kafka message.

        Raises:
            DataIntegrityException: If message is invalid
        """
        try:
            # Validation 1: Timestamp format validation
            created_at = _parse_iso_timestamp(message.created_at)
            inserted_at = _parse_iso_timestamp(message.inserted_at)

            # Validation 2: Business rule validation
            if not message.line_items:
                raise ValueError("Order must have at least one line item")

            if len(message.order_id) == 0:
                raise ValueError("Order ID cannot be empty")

            # Validation 3: Line item constraint validation
            line_items: list[OrderItem] = []
            for item in message.line_items:
                if item.quantity <= 0:
                    raise ValueError(f"Invalid quantity: {item.quantity}")
                if not item.product_title.strip():
                    raise ValueError("Product title cannot be empty")

                line_items.append(OrderItem(
                    line_item_id=item.line_item_id,
                    product_id=item.product_id,
                    product_title=item.product_title,
                    quantity=item.quantity
                ))

            # All validations passed - translate to domain entity
            order = Order(
                order_id=message.order_id,
                created_at=created_at,
                line_items=line_items,
                inserted_at=inserted_at
            )

            logger.info("order_translated", order_id=message.order_id)
            return order

        except ValueError as e:
            logger.error("order_translation_failed", error=str(e))
            raise DataIntegrityException(f"Invalid order data: {e}") from e


def _parse_iso_timestamp(timestamp_str: str) -> datetime:
    """Parse ISO 8601 timestamp with validation."""
    try:
        return datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
    except ValueError as e:
        raise ValueError(f"Invalid ISO timestamp: {timestamp_str}") from e
```

## Instructions

### Step 1: Understand the Defense-in-Depth Pattern

The skill is built on **validating at both producer and consumer layers**:

- **Producer Validation (Extraction Context):** Validate business rules and schema compliance before publishing to Kafka using msgspec schemas and domain value objects
- **Consumer Validation (Storage Context):** Validate data integrity and transformation when consuming from Kafka using anti-corruption layers
- **ClickHouse Error Streaming:** Capture malformed messages in a separate error table instead of blocking consumption with `kafka_handle_error_mode='stream'`
- **Idempotency:** Handle at-least-once delivery semantics using ReplacingMergeTree for automatic deduplication

### Step 2: Set Up