WKafka 🚀

Production-Grade, Decorator-Driven Apache Kafka Framework for Python.

WKafka simplifies event-driven architectures by eliminating boilerplate code. It transforms raw, low-level Apache Kafka clients into an intuitive, elegant, and resilient development experience. With first-class native support for JSON, YAML, Computer Vision Images (OpenCV / Pillow), Arbitrary Binary Files (PDF, ZIP, TXT), and Strict Pydantic Validation, WKafka powers high-throughput microservices, IoT telemetry, real-time multimedia, and AI/ML inference pipelines.

🌟 Why WKafka?

Raw Kafka clients require dozens of lines of repetitive boilerplate: manually serializing bytes, managing poll loops, handling heartbeat timeouts during heavy tasks, orchestrating thread pools, decoding headers, and configuring retries.

WKafka solves these challenges out of the box:

• 🎯 Clean Decorator Syntax: Register consumers with a single @kafka.consumer(...) annotation.
• 🧬 5 Native Serializers: Seamlessly transfer JSON, YAML, OpenCV/NumPy images, raw files, and Pydantic schemas without manual byte conversion.
• ⚡ Decoupled Heavy Computation: Prevent Kafka's infamous CommitFailedError / max_poll_interval_ms rebalance loops during hours-long ML tasks via background=True and commit_on_receive=True.
• 🛡️ Built-in Resilience: Automatic retries with exponential backoff and automated Dead Letter Queue (DLQ) routing with diagnostic error headers.
• 🚀 Dynamic Partition Auto-Scaling: Automatically detect consumer concurrency and scale topic partitions on the broker via KafkaAdminClient (partition_scale=True).
• ⚡ High-Throughput Payload Offloading: Stream high-FPS computer vision matrices or heavy files directly in Redis RAM with automated TTL via wredis while Kafka routes ultra-lightweight Claim-Check references (~120 bytes).
• 🔒 Enterprise Security: Native support for SASL (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512), SSL/TLS, and SOCKS5 proxies.
• ⚡ Async/Await Ready: Native execution of asynchronous coroutine handlers (async def).

📦 Installation

# Basic installation via pip
pip install wkafka

# Via poetry
poetry add wkafka

# With high-speed Redis payload offload support
pip install "wkafka[redis]"

# High-performance snappy compression support
pip install "wkafka[snappy]"

# All optional enterprise features
pip install "wkafka[all]"

🚀 5-Minute Quick Start

1. Minimal Producer & Consumer

from wkafka import WKafka

# Connects to KAFKA_SERVER environment variable or defaults to localhost:9092
kafka = WKafka(bootstrap_servers="localhost:9092")

# Register consumer handler
@kafka.consumer(topic="orders.created", format="json")
def on_order_created(msg):
    print(f"📦 [Consumer] Order received: {msg.value}")
    print(f"   Topic: {msg.topic}, Partition Offset: {msg.offset}")

# Start consumers in a managed thread pool
kafka.run_consumers(block=False)

# Produce message safely within a context manager
with kafka.producer() as p:
    p.send(
        topic="orders.created",
        value={"order_id": "ORD-9842", "customer": "Alice", "total": 149.99},
        format="json",
    )

🧬 Built-in Serializers Deep Dive

WKafka includes 5 specialized serializers. No manual .encode("utf-8"), json.loads(), or image decoding is required:

1. JSON (format="json")

Serializes standard Python dictionaries, lists, primitives, and nested objects.

with kafka.producer() as p:
    p.send("iot.telemetry", value={"device_id": "sensor-12", "temp": 23.5}, format="json")

@kafka.consumer(topic="iot.telemetry", format="json")
def handle_telemetry(msg):
    print(f"Temperature: {msg.value['temp']} °C")

2. YAML (format="yaml")

Ideal for configuration payloads, declarative job descriptions, and Kubernetes-style manifests.

job_config = {
    "version": "v1",
    "pipeline": "face_clustering",
    "resources": {"gpu": 1, "memory": "8Gi"}
}
with kafka.producer() as p:
    p.send("jobs.deploy", value=job_config, format="yaml")

@kafka.consumer(topic="jobs.deploy", format="yaml")
def handle_job(msg):
    # msg.value is deserialized back into a Python dict via PyYAML safe_load
    print(f"Pipeline: {msg.value['pipeline']}")

3. Images & Video Frames (format="image")

Native integration with OpenCV (cv2), NumPy, and Pillow (PIL.Image). Produces compressed JPEG byte streams and deserializes directly into OpenCV BGR numpy arrays (np.ndarray) ready for deep learning models.

import cv2
import numpy as np
from PIL import Image

# Produce an OpenCV BGR frame or Pillow Image
frame = cv2.imread("frame_001.jpg")  # np.ndarray
with kafka.producer() as p:
    # Optional 'quality' parameter (1-100, default: 80)
    p.send("camera.stream", value=frame, format="image", quality=90)

@kafka.consumer(topic="camera.stream", format="image")
def process_frame(msg):
    # msg.value is a decoded cv2 BGR numpy array ready for inference
    image_bgr = msg.value
    h, w, c = image_bgr.shape
    print(f"📸 Frame received: {w}x{h}, {c} channels")

4. Arbitrary Binary Files & Documents (format="file")

Transmit entire PDF reports, ZIP archives, models, audio, or text files directly. You can pass:

• A local file system path string (e.g. "/path/to/report.pdf").
• Raw bytes.
• A dictionary containing a "content" key with bytes.

# Producer sends a PDF document directly by file path
with kafka.producer() as p:
    p.send("reports.generated", value="/tmp/quarterly_audit.pdf", format="file")

@kafka.consumer(topic="reports.generated", format="file")
def handle_pdf(msg):
    pdf_bytes = msg.value  # Raw binary bytes
    with open("/storage/received_audit.pdf", "wb") as f:
        f.write(pdf_bytes)
    print(f"📄 Saved PDF file ({len(pdf_bytes)} bytes)")

5. Pydantic Models (format="pydantic")

Strict typed validation for both Pydantic v1 and v2. Guarantees that consumer handlers only receive valid, validated domain models.

from pydantic import BaseModel, Field

class UserEvent(BaseModel):
    user_id: int
    email: str
    active: bool = True

# Producer sends Pydantic model instance
event = UserEvent(user_id=101, email="wisrovi@example.com")
with kafka.producer() as p:
    p.send("users.events", value=event, format="pydantic")

# Consumer automatically validates and parses back into the model class
@kafka.consumer(topic="users.events", format="pydantic", model=UserEvent)
def handle_user(msg):
    user: UserEvent = msg.value
    print(f"👤 User: {user.user_id}, Email: {user.email}")

📤 Producer Guide

Context Manager & Safe Flushing

The with kafka.producer() as p: pattern guarantees that messages buffered in memory are completely flushed (p.flush()) to the Kafka broker and sockets are cleanly closed even if exceptions occur.

with kafka.producer() as p:
    for i in range(10):
        p.send("metrics", value={"cpu": i * 10}, format="json")
# Guaranteed flush & close upon exiting the block

Message Keys & Partition Routing

Provide key="entity_id" to ensure all events with the same key land on the same partition, guaranteeing strict chronological ordering.

with kafka.producer() as p:
    p.send("sensor.data", key="sensor-alpha", value={"reading": 42.1}, format="json")

Headers & Metadata

Pass rich metadata dictionaries via headers. WKafka automatically serializes dictionary and list headers into JSON bytes, and automatically parses them back on the receiving consumer end:

headers = {
    "trace_id": "7b8f9e2d-3c4a-4e89",
    "tenant_id": 502,
    "routing_info": {"datacenter": "eu-central-1", "priority": "high"},
}

with kafka.producer() as p:
    p.send("orders", value={"order": 1}, headers=headers, format="json")

High-Throughput Snappy Compression

If python-snappy is installed (pip install "wkafka[snappy]"), WKafka automatically enables Snappy compression for maximum throughput and minimal network bandwidth. If not installed, it gracefully falls back to gzip compression with an informative log.

📥 Consumer Guide

1. Manual Offset Commits (At-Least-Once Delivery)

By default, Kafka uses auto-commit. When you require strict At-Least-Once guarantees where an offset must only be marked as completed after successful database persistence, set auto_commit=False and call msg.commit():

@kafka.consumer(topic="payments", format="json", auto_commit=False)
def process_payment(msg):
    try:
        charge_credit_card(msg.value)
        record_in_database(msg.value)
        # Commit offset only after business operations succeed
        msg.commit()
    except DatabaseConnectionError:
        # Message is not committed; will be safely re-consumed
        raise

2. Automatic Retries & Dead Letter Queue (DLQ)

When external services fail temporarily, configure retries with exponential backoff. If all attempts fail, the message is automatically routed to a Dead Letter Queue topic with diagnostic headers:

@kafka.consumer(
    topic="external.sync",
    format="json",
    max_retries=3,          # Retry up to 3 times
    retry_delay=1.0,        # Delays: 1s, 2s, 4s (exponential)
    dlq_topic="external.sync.DLQ"
)
def sync_handler(msg):
    call_flaky_third_party_api(msg.value)

Failed messages routed to dlq_topic automatically receive the following headers:

• x-error-message: Exact Python exception text that caused the final failure.
• x-original-topic: Source topic where the message originated.
• x-original-offset: Source partition offset.

3. Multi-Topic & Regex Pattern Subscriptions

Subscribe to multiple explicit topics or dynamically match topics using regular expressions:

# Multiple explicit topics
@kafka.consumer(topic=["telemetry.eu", "telemetry.us"], format="json")
def handle_global_telemetry(msg):
    print(f"Data from {msg.topic}: {msg.value}")

# Dynamic Regex pattern matching
@kafka.consumer(pattern="iot\\..*\\.sensors", format="json")
def handle_dynamic_sensors(msg):
    print(f"Regex matched event on {msg.topic}: {msg.value}")

4. Partition Key Filtering

Filter incoming messages by partition key directly at the decorator level, ignoring unrelated keys without business logic overhead:

@kafka.consumer(topic="logs", format="json", key_filter="CRITICAL")
def handle_critical_logs_only(msg):
    send_pagerduty_alert(msg.value)

5. Asynchronous Coroutines (async / await)

Handlers can be native Python coroutines. WKafka manages event loops and thread safety transparently:

import aiohttp

@kafka.consumer(topic="webhooks.dispatch", format="json")
async def dispatch_webhook(msg):
    async with aiohttp.ClientSession() as session:
        async with session.post(msg.value["url"], json=msg.value["payload"]) as resp:
            print(f"Webhook response status: {resp.status}")

⚡ Long-Running Tasks & Decoupled Workers (The ML Dilemma)

The Problem: Heartbeat Timeouts & Rebalance Storms

In Apache Kafka, if a consumer handler takes longer than max_poll_interval_ms (default: 5 minutes) to return:

1. The Kafka broker assumes the consumer has crashed or deadlocked.
2. The broker evicts the consumer from the consumer group (Consumer Group Rebalance).
3. The message offset is never committed, and another consumer picks up the same message in an infinite duplicate loop.

The Solution: WKafka Decoupled Architecture

WKafka provides two native solutions that can be used independently or combined:

1. commit_on_receive=True: Immediately commits the partition offset to the broker the moment the message is dequeued, before running heavy processing logic.
2. background=True: Immediately submits the message to an internal ThreadPoolExecutor. The main Kafka polling thread instantly returns to its poll() loop, continually answering broker heartbeats. Heavy ML model inference, large video rendering, or PDF generation can run for hours without triggering a rebalance!
3. background_workers=N: Controls the maximum number of concurrent tasks executed in parallel.

from wkafka import WKafka
import time

kafka = WKafka(bootstrap_servers="localhost:9092")

# Long-running ML inference or heavy document generation
@kafka.consumer(
    topic="ai.document.ocr",
    format="json",
    commit_on_receive=True,  # 1. Acknowledge offset immediately upon receipt
    background=True,         # 2. Offload to background worker pool
    background_workers=4,    # 3. Allow up to 4 parallel heavy jobs
)
def heavy_ocr_pipeline(msg):
    # Runs in an isolated thread.
    # The Kafka consumer poll loop remains active, sending heartbeats continuously!
    print(f"🚀 Processing document {msg.value['doc_id']}...")
    time.sleep(600)  # Simulating a 10-minute AI task
    print(f"✅ Finished document {msg.value['doc_id']}")

kafka.run_consumers(block=True)

Architecture Diagram

Kafka Topic ────────► KafkaConsumer Loop Thread (poll)
                            │
                            ├─► commit_on_receive=True ──► [Immediate Broker Offset Commit]
                            │
                            └─► background=True ─────────► [ThreadPoolExecutor]
                                                               ├─► Worker Thread 1 (Heavy ML / PDF)
                                                               ├─► Worker Thread 2 (Heavy ML / PDF)
                                                               └─► Polling Thread Continues (Heartbeats Active!)

🚀 Dynamic Partition Auto-Scaling (partition_scale=True)

In Apache Kafka, the number of parallel consumers in a consumer group cannot exceed the number of partitions on the topic. If a topic has 1 partition and you run 4 consumer threads, 3 threads remain completely idle.

When partition_scale=True is enabled in WKafka() or run_consumers(partition_scale=True):

1. WKafka counts the number of registered consumer handlers for each topic.
2. It queries the broker via KafkaAdminClient.describe_topics().
3. If the topic exists with fewer partitions than required, it dynamically scales the partition count via KafkaAdminClient.create_partitions().
4. If the topic does not exist, it automatically creates it with the desired partition count.
5. All worker threads can now consume partitions in true parallel fashion!

kafka = WKafka(bootstrap_servers="localhost:9092", partition_scale=True)

🚀 High-Throughput RAM Payload Offloading (Claim-Check Pattern with WRedis)

The Vision & Large-File Bottleneck

Streaming high-frequency multimedia (e.g. 30 FPS at 1080p OpenCV matrices) or multi-megabyte binary documents directly through Apache Kafka brokers leads to severe problems:

1. Broker Memory & Disk Saturation: Heavy messages bloat the JVM heap, trigger garbage collection stalls, and monopolize disk pagecaches.
2. Commit Latency & Rebalances: Transporting heavy binaries slows down consumer polling and offset commits.
3. Stale Frames Accumulation: If a downstream consumer lags during deep learning inference (e.g., YOLO or segmentation), Kafka queues up gigabytes of outdated frames instead of maintaining real-time video state.

The Solution: Transparent Claim-Check via WRedis

WKafka natively implements the Claim-Check Pattern powered by WRedis:

1. Producer Side: Heavy payloads (e.g. format="image" or format="file") are compressed and written directly to Redis in RAM with a short Time-To-Live (TTL = 5 seconds).
2. Kafka Broker: Transfers only an ultra-lightweight JSON Claim-Check pointer (~120 bytes) containing headers and routing keys.
3. Consumer Side: Intercepts the Claim-Check reference, retrieves the binary payload directly from Redis RAM, and decodes it back to the original object (numpy.ndarray for images) transparently.
4. Drop-on-Expired Policy: If a consumer lags beyond the TTL (5s), stale frames evaporate from RAM and are skipped automatically without crashing (on_expired="drop"), keeping your pipeline strictly real-time!

sequenceDiagram
    autonumber
    actor P as Producer (Camera Extractor)
    participant WK as WKafka Engine
    participant R as Redis RAM (WRedis)
    participant K as Kafka Broker
    actor C as Consumer (Vision AI / YOLO)

    P->>WK: send("camera-stream", frame, format="image")
    Note over WK: Serializes frame to JPEG bytes.<br/>Detects offload active
    WK->>R: SETEX("wkafka:claim:<uuid>", 5s, raw_bytes)
    WK->>K: send(Topic, Header & JSON pointer: ~120 bytes)
    Note over K: Broker handles sub-millisecond routing
    K->>WK: raw_message
    Note over WK: Intercepts x-wkafka-offload header
    WK->>R: GET("wkafka:claim:<uuid>")
    alt Frame alive in RAM (within 5s TTL)
        R-->>WK: raw_bytes
        Note over WK: Deserializes to np.ndarray
        WK->>C: callback(msg) with original frame
    else Frame expired (> 5s consumer lag)
        Note over WK: on_expired="drop": Drops stale frame safely
        WK-->>C: Skips callback, preserves 0 lag
    end

Flexible Configuration (Options A, B, C, & D)

Option A: Load from Full Application YAML (config.yaml)

Ideal for Kubernetes ConfigMaps where infrastructure settings are updated without altering Python code:

kafka:
  bootstrap_servers: "192.168.1.68:9092"
  client_id: "vision-pipeline-extended"
  offload:
    backend: "redis"
    host: "192.168.1.68"
    port: 15963
    password: null
    db: 0
    ttl: 5
    on_expired: "drop"
    formats:
      - "image"
      - "file"
    threshold_kb: 50
    key_prefix: "wkafka:claim:camera_1"

In Python:

kafka = WKafka(config="config.yaml")
# Or directly as first positional argument:
kafka = WKafka("config.yaml")

Option B: Load Offload Block from Dedicated YAML

kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload="offload_config.yaml"
)

Option C: In-Memory Dictionary (Minimal or Extended)

# Minimal configuration (uses defaults and localhost:6379 / REDIS_HOST)
kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload={"backend": "redis", "ttl": 5}
)

# Extended configuration with custom network host/port and drop policy
kafka = WKafka(
    bootstrap_servers="localhost:9092",
    offload={
        "backend": "redis",
        "host": "192.168.1.68",
        "port": 15963,
        "password": "<your_redis_password>",
        "db": 0,
        "ttl": 5,
        "on_expired": "drop",
        "formats": ["image", "file"],
        "threshold_kb": 50,
        "key_prefix": "wkafka:claim:cam1",
    }
)

Option D: Environment Variables (Docker & Kubernetes)

export REDIS_HOST="192.168.1.68"
export REDIS_PORT="15963"

kafka = WKafka(offload={"backend": "redis", "ttl": 5})

Producer & Consumer Implementation (100% Transparent)

# 1. Producer (Camera Extractor)
kafka = WKafka(bootstrap_servers="localhost:9092", offload={"backend": "redis", "ttl": 5})

# Send as usual - offloading is transparent:
kafka.send("camera-stream", value=frame, format="image")

# 2. Consumer (Inference / Analytics)
@kafka.consumer("camera-stream", format="image")
def process_frame(msg):
    frame = msg.value  # Seamlessly receives the numpy.ndarray from Redis RAM

🔒 Enterprise Security & Proxies

SASL Authentication (PLAIN / SCRAM)

WKafka seamlessly passes authentication credentials down to KafkaConsumer, KafkaProducer, and KafkaAdminClient:

sasl_config = {
    "security_protocol": "SASL_PLAINTEXT",  # or SASL_SSL
    "sasl_mechanism": "SCRAM-SHA-512",      # PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
    "sasl_plain_username": "enterprise-service",
    "sasl_plain_password": "super-secure-password",
}

kafka = WKafka(
    bootstrap_servers="kafka-secure.internal:9093",
    **sasl_config
)

SOCKS5 Proxy Normalization

If operating behind corporate firewalls or SOCKS proxies, WKafka automatically normalizes deprecated proxy settings (socks5_proxy to proxy_url) without triggering client warnings.

📋 Comprehensive API Reference

1. WKafka Constructor Reference

WKafka(bootstrap_servers=None, client_id=None, dynamic_group_id=False, partition_scale=False, **extra_config)

2. @kafka.consumer(...) Decorator Reference

@kafka.consumer(topic=None, pattern=None, group_id=None, format="json", auto_commit=True, commit_on_receive=False, background=False, background_workers=1, max_retries=0, retry_delay=1.0, dlq_topic=None, model=None, key_filter=None, **kafka_kwargs)

3. Producer Reference (kafka.send(...))

kafka.send(topic, value, key=None, format="json", headers=None, **kwargs)

4. Message Object Reference

Received by consumer callback functions:

🗂️ 15 Production-Ready Examples

WKafka includes a comprehensive catalog of 15 fully functional example modules in the examples/ directory:

🛠️ Key Technologies and Libraries

• Python (3.9 - 3.14): Core programming language and execution runtime.
• kafka-python-ng / kafka-python: Low-level client driver for Apache Kafka and KRaft protocol communication.
• WRedis (wredis): High-speed in-memory database and caching layer for ephemeral multimedia offloading.
• OpenCV (opencv-python) & Pillow: Computer vision matrix manipulation, image decoding, rendering, and JPEG compression.
• NumPy: Multidimensional arrays and tensor buffers for real-time video matrices.
• Pydantic (v1 / v2): Declarative validation and strictly-typed data contracts.
• PyYAML: Safe parsing and serialization of declarative configuration manifests.
• Loguru: Structured logging framework for distributed event traceability.
• Pytest & Coverage: Automated unit test suites and comprehensive coverage reporting.
• Docker: Isolated sandboxing for consistent testing across environments.

🧪 Unit Testing & Coverage

The project includes an exhaustive unit test suite verifying serializers, retry policies, DLQ routing, manual offsets, decoupled execution, dynamic partition scaling, and Claim-Check payload offload.

1. Run Tests Locally (pytest)

poetry run pytest tests/ -v

2. Generate Coverage Report (Guaranteed >80%)

./run_coverage.sh

This script runs pytest --cov=wkafka --cov-fail-under=80 and outputs a comprehensive HTML report in htmlcov/index.html. It currently guarantees over 88% code coverage.

3. Run Tests in Isolated Docker Container

To ensure strict parity with CI/CD and production environments:

./run_tests_docker.sh

The script runs the full test suite inside an isolated python:3.13-slim container with all operating system and Python dependencies installed.

👤 Author & Affiliation

• Author: William Steve Rodriguez Villamizar (Wisrovi)
• Title: AI Leader & Solutions Architect at eCaptureDtech
• Location: Badajoz, Extremadura, Spain
• 🌐 Official Website: wisrovi.dev
• 💼 LinkedIn: wisrovi-rodriguez
• 🆔 ORCID: 0009-0005-0710-1861
• 📦 PyPI: pypi.org/user/wisrovi/
• 🐙 GitHub: @wisrovi

📜 License

Distributed under the MIT License. Free for commercial, enterprise, and academic research use.