glean-klient 0.1.1

Add your description here

klient

Lightweight Python wrappers around confluent-kafka providing:

• Unified Producer, Consumer, and Admin helpers with clear sync & async APIs
• Built-in transactional produce (sync & async context managers)
• Environment–aware configuration loading from ~/.kafka (merge default + named env)
• Simple CLI entrypoint (python -m klient ...) for common admin & consume patterns
• Sensible defaults (isolation level defaults to read_committed, unlimited streaming)

The project favors clarity over cleverness: thin abstractions, explicit naming, and test-backed behavior.

---

Installation

Prerequisites

Kafka broker accessible (local or remote), Python >=3.12.

Install via editable clone

git clone https://github.com/your-org/glean-kafka.git
cd glean-kafka
pip install -e .[dev]

The [dev] extra (defined in pyproject.toml) installs testing dependencies (pytest, pytest-asyncio, pytest-cov).

---

Quick Start (Library)

Producing Messages

from klient.producer import KafkaProducer, ProducerConfig

producer = KafkaProducer(ProducerConfig(bootstrap_servers="localhost:9092"))
result = producer.produce(topic="events", key=b"user-1", value=b"hello")
print(result.status, result.partition, result.offset)

# Async
import asyncio
async def main():
    aresult = await producer.aproduce(topic="events", key=b"user-2", value=b"hi async")
    print(aresult.status)
asyncio.run(main())

producer.close()

Consuming (poll single / batch / stream)

from klient.consumer import KafkaConsumer, ConsumerConfig

consumer = KafkaConsumer(ConsumerConfig(bootstrap_servers="localhost:9092", group_id="demo", topics=["events"]))  # supply topics list in config

# Explicit subscription (list-only API). Provide a list even for a single topic:
consumer.subscribe(["events"])  # list of topic strings

# Single poll (returns MessageResult | None)
msg = consumer.poll(timeout=1.0)
if msg:
    print(msg.key, msg.value)

# Batch consume (up to max_messages or until timeout window exhausted)
batch = consumer.consume_messages(max_messages=10, timeout=5.0)
for m in batch:
    print(m.offset)

# Streaming (unlimited by default). Break manually.
for m in consumer.message_stream(timeout=1.0):
    print(m.key, m.value)
    if some_condition():
        break
consumer.stop()  # stop underlying loop if streaming helper created one

Seeking to Specific Offsets

from klient.consumer import KafkaConsumer, ConsumerConfig

consumer = KafkaConsumer(ConsumerConfig(bootstrap_servers="localhost:9092", group_id="demo"))
consumer.subscribe(["events"])

# Seek to offset 1000 in ALL partitions of the topic
consumer.seek_to_offset(topic="events", offset=1000)

# Seek to offset 1000 in a SPECIFIC partition
consumer.seek_to_offset(topic="events", offset=1000, partition=0)

# Now poll or consume from that offset
msg = consumer.poll(timeout=1.0)
if msg:
    print(f"Message at offset {msg.offset}: {msg.value}")

Admin Operations

from klient.admin import KafkaAdmin, AdminConfig

admin = KafkaAdmin(AdminConfig(bootstrap_servers="localhost:9092"))
admin.create_topics([{ "topic": "events", "num_partitions": 1, "replication_factor": 1 }])
print(admin.list_topics())
admin.delete_topics(["events"])  # cleanup

---

Transactions

Enable transactions by supplying a transactional_id in ProducerConfig. The wrapper initializes transactions lazily.

from klient.producer import KafkaProducer, ProducerConfig

producer = KafkaProducer(ProducerConfig(bootstrap_servers="localhost:9092", transactional_id="tx-demo"))

with producer.transaction():  # sync context manager
    producer.produce("events", key=b"k1", value=b"payload-1")
    producer.produce("events", key=b"k2", value=b"payload-2")
    # raise to trigger abort

# Async
import asyncio
async def run_tx():
    async with producer.atransaction():
        await producer.aproduce("events", key=b"k3", value=b"async-1")
asyncio.run(run_tx())

Error handling: a raised exception inside the context causes abort_transaction() and re-raises. Begin/commit/abort each return a TransactionResult with success, error, and optional transaction_id.

Continuous Async Transaction Stream

Below is an example of producing messages continuously in discrete transactional batches. Each loop iteration creates a new transaction, allowing partial failure without impacting previous committed batches. A cancellation signal (Ctrl+C) or external event stops the loop gracefully.

import asyncio
import signal
from klient.producer import KafkaProducer, ProducerConfig

stop = asyncio.Event()

def handle_sigint(*_):
  stop.set()

signal.signal(signal.SIGINT, handle_sigint)

producer = KafkaProducer(ProducerConfig(
  bootstrap_servers="localhost:9092",
  transactional_id="stream-tx-producer"  # stable id per producer instance
))

async def produce_forever(batch_size: int = 5, delay: float = 0.5):
  counter = 0
  while not stop.is_set():
    async with producer.atransaction():  # new transaction each batch
      for i in range(batch_size):
        key = f"user-{counter}".encode()
        value = f"payload-{counter}".encode()
        await producer.aproduce("events", key=key, value=value)
        counter += 1
    await asyncio.sleep(delay)  # pacing; remove for max throughput

  # Optional final flush (transactions already committed)
  producer.flush()

asyncio.run(produce_forever())

Considerations:

1. Throughput: Increase batch_size and reduce delay for higher message rates.
2. Ordering: All messages in a single transaction appear atomically to read_committed consumers.
3. Backpressure: If delivery reports back up, adjust producer config (e.g., linger, batch size, queue limits).
4. Shutdown: SIGINT sets the event; the current transaction completes before exit.
5. Retry semantics: confluent-kafka handles retries internally; aborted transactions never become visible when isolation is read_committed.

Consuming From One Topic and Producing To Another (Relay)

Common pattern: read, transform, and forward. Below are sync and async relay examples. The async version batches each relay group into a transaction for atomicity.

Synchronous Relay (Simple Transform)

from klient.consumer import KafkaConsumer, ConsumerConfig
from klient.producer import KafkaProducer, ProducerConfig

consumer = KafkaConsumer(ConsumerConfig(
  bootstrap_servers="localhost:9092",
  group_id="relay-group",
  isolation_level="read_committed"
))
consumer.subscribe(["input-topic"])  # list-only subscription

producer = KafkaProducer(ProducerConfig(
  bootstrap_servers="localhost:9092"
))

for msg in consumer.message_stream(timeout=1.0):
  # Basic transform: uppercase value, preserve key
  new_value = msg.value.upper() if msg.value else b""
  producer.produce("output-topic", key=msg.key, value=new_value)
  # Optionally flush periodically for latency control
  # if msg.offset % 100 == 0: producer.flush()

Async Transactional Relay (Batch Atomicity)

import asyncio
from klient.consumer import KafkaConsumer, ConsumerConfig
from klient.producer import KafkaProducer, ProducerConfig

consumer = KafkaConsumer(ConsumerConfig(
  bootstrap_servers="localhost:9092",
  group_id="relay-async-group",
  isolation_level="read_committed"
))
consumer.subscribe(["input-topic"])  # list-only subscription

producer = KafkaProducer(ProducerConfig(
  bootstrap_servers="localhost:9092",
  transactional_id="relay-tx-producer"
))

async def relay_forever(batch_size: int = 25):
  buffer = []
  async for msg in consumer.amessage_stream(timeout=1.0):
    buffer.append(msg)
    if len(buffer) >= batch_size:
      async with producer.atransaction():
        for m in buffer:
          # Example enrichment: append offset metadata
          enriched = (m.value or b"") + f"|offset={m.offset}".encode()
          await producer.aproduce("output-topic", key=m.key, value=enriched)
      buffer.clear()

asyncio.run(relay_forever())

Notes:

1. Backpressure: Adjust batch_size to tune commit frequency vs latency.
2. Ordering: Within a transaction all output messages become visible together; cross-transaction ordering depends on source order + processing time.
3. Isolation: Using read_committed avoids relaying aborted input messages.
4. Flow Control: Add a max loop runtime or cancellation signal for graceful shutdown.
5. Error Handling: Exceptions inside the transactional context abort that batch only; upstream offsets still advance because messages were read—consider manual offset management if exactly-once relay semantics are required.

---

Environment Configuration Loading

Single file model: ~/.kafka/config.json (or an explicit path you pass). This JSON file may contain:

• A default object with baseline Kafka client properties.
• One or more named environment objects (dev, prod, etc.).

When an environment name is provided (CLI --env or wrapper factory parameter), the effective configuration is the shallow merge of default overlaid by the named environment object (environment values win). If no environment is specified, the default object is used; if default is absent, the raw top-level mapping is returned.

No per-environment standalone files are read; only the single config file is considered.

Role-Specific Environment Names

Producer and consumer often require different Kafka client properties (e.g. batching, compression, fetch settings). Define separate environment blocks like prod-producer and prod-consumer in the same config file:

{
  "default": {"bootstrap.servers": "shared:9092", "client.id": "app-base"},
  "prod-producer": {"bootstrap.servers": "prod-write:9092", "compression.type": "lz4", "linger.ms": 25},
  "prod-consumer": {"bootstrap.servers": "prod-read:9092", "auto.offset.reset": "earliest", "fetch.max.bytes": 5242880}
}

Library usage:

from klient import resolve_env_config, KafkaProducer, ProducerConfig, KafkaConsumer, ConsumerConfig

producer_raw = resolve_env_config('prod-producer', None)
consumer_raw = resolve_env_config('prod-consumer', None)

producer = KafkaProducer(ProducerConfig(
    bootstrap_servers=producer_raw['bootstrap.servers'],
    additional_config={k: v for k, v in producer_raw.items() if k != 'bootstrap.servers'}
))

consumer = KafkaConsumer(ConsumerConfig(
    bootstrap_servers=consumer_raw['bootstrap.servers'],
    group_id='analytics-group',
    additional_config={k: v for k, v in consumer_raw.items() if k not in ('bootstrap.servers', 'group.id')}
))
consumer.subscribe(['input-topic'])

CLI usage with role-specific environments:

python -m klient --config-file ~/.kafka/config.json \
  --producer-env prod-producer \
  --consumer-env prod-consumer \
  produce transaction events --count 10 --transactional-id tx-prod

python -m klient --config-file ~/.kafka/config.json \
  --consumer-env prod-consumer \
  consume stream events --timeout 1

If a role-specific env is not provided, the global --env (if set) applies; otherwise only default keys are used.

Section splitting: A combined env file may contain scoped objects:

{
  "default": {
    "bootstrap_servers": "localhost:9092"
  },
  "dev": {
    "bootstrap_servers": "dev-broker:9092",
    "producer": {"linger_ms": 5},
    "consumer": {"group_id": "demo-group", "auto_offset_reset": "earliest"},
    "admin": {"security_protocol": "PLAINTEXT"}
  }
}

Factory helpers resolve and split config (note: subscribe now requires a list):

from klient.producer import KafkaProducer
prod = KafkaProducer.from_env_config("dev")

from klient.consumer import KafkaConsumer
cons = KafkaConsumer.from_env_config("dev", topics=["events"])  # topics is list

from klient.admin import KafkaAdmin
adm = KafkaAdmin.from_env_config("dev")

If an env is missing, an empty dict is returned (wrappers fall back to explicit arguments or defaults like localhost:9092).

---

Isolation Level (Default: read_committed)

The consumer default was changed from read_uncommitted to read_committed to hide aborted transactional messages. Override per CLI flag or ConsumerConfig(isolation_level="read_uncommitted") when debugging.

Level	Behavior
read_committed	Only committed transactional messages + non-transactional
read_uncommitted	Includes pending + aborted transactional messages

---

CLI Usage

Invoke via module:

python -m klient --help

Configuration Display

Use --show-config to display all Kafka configuration details before executing any command:

# Show configuration for a consumer command
python -m klient --show-config consume poll events --group my-group

# Show configuration with specific environment
python -m klient -e production --show-config admin list-topics

# Show configuration for producer with detailed command info
python -m klient --show-config produce send events --key user1 --value "hello"

The configuration display includes:

• Connection Details: Bootstrap servers, environment names
• Command Information: Specific command parameters and options
• Configuration Sections: All producer, consumer, and admin settings
• Security: Sensitive values (passwords, keys, secrets) are automatically masked
• Available Environments: Lists all configured environments from config file

Output Options

Control output formatting, filtering, and destination:

# Write output to file instead of stdout
python -m klient admin list-topics --output-file topics.json

# Pretty-formatted JSON (default)
python -m klient --json info config-dump

# Compact JSON (single line, no extra spaces)
python -m klient --json --compact-json info config-dump

# Filter output with regex pattern
python -m klient --filter "prod" info config-dump

# Filter JSON by specific key (key:regex syntax)
python -m klient --json --filter "bootstrap.servers:kafka" info config-dump

# Combine all options: filtered, formatted output to file
python -m klient --json --compact-json --filter "server:prod" --output-file filtered.json info config-dump

Filtering Examples

Text Filtering:

# Show only lines containing "kafka"
python -m klient --filter "kafka" admin list-topics

# Case-insensitive regex patterns
python -m klient --filter "prod.*server" info config-dump

JSON Key Filtering:

# Filter by specific key values
python -m klient --json --filter "topic:events" consume poll events

# Filter nested JSON structures  
python -m klient --json --filter "host:prod" info config-dump

# Use regex patterns in key filtering
python -m klient --json --filter "name:^test.*" admin list-topics

Key commands (abbreviated) reflecting current subcommand syntax:

Produce one message (sync):

python -m klient produce send events --key k1 --value hello

Transactional batch (N messages in one commit):

python -m klient produce transaction events --count 5 --transactional-id batch-1

Produce from JSON file (array format: [{}, {}, ...]):

# Basic usage - send entire JSON object as message value
python -m klient produce from-file events messages.json

# Extract specific fields for message components
python -m klient produce from-file events data.json --key-field user_id --partition-field shard

# Transactional batch processing from file
python -m klient produce from-file events batch.json --transactional-id tx-1 --batch-size 50

# Extract headers from JSON field
python -m klient produce from-file events events.json --headers-field metadata --key-field id

JSON File Format: The JSON file must contain an array of objects. Each object becomes a separate Kafka message:

[
  {
    "user_id": "user1", 
    "data": "message content",
    "shard": 0,
    "metadata": {"type": "event", "version": "1.0"}
  },
  {
    "user_id": "user2",
    "data": "another message", 
    "shard": 1,
    "metadata": {"type": "command", "version": "2.0"}
  }
]

Field extraction options:

• --key-field: Extract this field as message key (removed from value)
• --partition-field: Extract this field as partition number (removed from value)
• --headers-field: Extract this object field as message headers (removed from value)
• Remaining fields become the JSON message value

Consuming Messages

All consume commands require a topic name as the first positional argument after the subcommand:

python -m klient consume <subcommand> <TOPIC> [options...]
# Examples:
python -m klient consume poll my-topic --group my-group
python -m klient consume batch events --count 10
python -m klient consume stream logs --limit 100

Poll once (single message attempt):

python -m klient consume poll events --group g1 --timeout 2

Batch (bounded fetch up to --count messages):

python -m klient consume batch events --group g1 --count 10 --timeout 5

Stream (unlimited; interrupt with Ctrl+C):

python -m klient consume stream events --group g1 --timeout 1

Stream with limit (first N then exit):

python -m klient consume stream events --group g1 --timeout 1 --limit 100

Stream with graceful shutdown grace period (allow in-flight processing to finish):

python -m klient consume stream events --group g1 --timeout 1 --grace-period 2.5

Seek to specific offset before consuming (available in poll, batch, and stream):

# Seek to offset 1000 in ALL partitions
python -m klient consume poll events --group g1 --seek-to-offset 1000

# Seek to offset 1000 in SPECIFIC partition
python -m klient consume poll events --group g1 --seek-to-offset 1000 --seek-to-partition 0

# Works with batch and stream commands too
python -m klient consume batch events --group g1 --seek-to-offset 500
python -m klient consume stream events --group g1 --seek-to-offset 2000 --seek-to-partition 1

Isolation override example:

python -m klient consume poll events --group g1 --isolation read_uncommitted

Config dump (merged view for env):

python -m klient info config-dump --env dev

See inline --help for each subcommand; consume help text explains poll vs batch vs stream semantics. Multi-topic subscription is supported only via library usage (consumer.subscribe(["t1","t2"])); the CLI stream/poll/batch commands accept a single topic argument.

---

Testing & Coverage

Run tests:

pytest -q

Collect coverage (library files; CLI omitted):

pytest --cov=klient --cov-report=term-missing

Goal: maintain >=80% coverage. Add focused tests for error paths before broad refactors.

---

Coding Guidelines

See .github/copilot-instructions.md for naming, error handling, logging, and testing standards. Changes affecting public API must update README + tests in the same commit.

---

Roadmap / Next Steps

• Sustain >=80% coverage; expand edge-path tests (rebalance failures, half-open circuit breaker)
• Add optional SASL/SSL config examples
• Structured logging integration hook for external log routers
• Benchmarks for high-throughput produce/consume scenarios
• Prometheus / OpenMetrics exporter (wrapping metrics.snapshot())
• Advanced circuit breaker (time-based half-open state)
• Pluggable metrics sink (push counters to external system)

---

Resiliency & Advanced Features

Error Classes

Producer:

• KafkaProducerError (base)
• KafkaProducerRetriableError (temporary issues; auto-retried in produce_with_retry)
• KafkaProducerFatalError (non-recoverable)
• KafkaTransactionError (transaction lifecycle failure)

Retry, Jitter & Circuit Breaker

produce_with_retry / aproduce_with_retry implement exponential backoff with +/-10% jitter to reduce thundering herds against brokers during partial outages. Backoff formula per attempt sleep = base_backoff * 2^(attempt-1) +/- 10% jitter.

Circuit breaker semantics: when the maximum attempts are exhausted for a single produce call, the breaker is considered "open" for that invocation (logged as JSON event produce_circuit_open / produce_async_circuit_open). A subsequent successful produce call logs circuit closure. This lightweight breaker prevents silent spin when persistent failures occur and provides an instrumentation hook (log monitor can alert on open events). For more advanced scenarios, extend with time-based half-open states.

Error Code Classification

Producer maintains sets of Kafka error codes for retriable, fatal, and fencing conditions (see producer.py: RETRIABLE_ERROR_CODES, FATAL_ERROR_CODES, FENCING_ERROR_CODES). These are populated defensively (wrapped in try/except AttributeError for version portability). Delivery callback logs classification buckets. Extend sets as operational patterns emerge.

Structured JSON Logging

Transaction lifecycle and retry circuit breaker events emit compact JSON objects (e.g. {"event":"transaction_commit","transaction_id":"relay-tx-producer","duration_ms":3.2}). Rebalance callbacks similarly emit {"event":"rebalance_assign",...} forms. This enables direct ingestion by log pipelines (Splunk/ELK). Avoid parsing non-JSON human-formatted logs.

Metrics (In-Process Counters)

Module klient.metrics offers thread-safe counters: inc(name: str, value: int = 1), get(name: str), and snapshot() (returns copy of all counters). Integrated / emitted counters:

• producer.tx.begin|commit|abort
• consumer.rebalance.assign|revoke|lost
• consumer.shutdown.signal (first SIGINT/SIGTERM received during stream)
• consumer.shutdown.complete (stream termination path executed)
• (Optional extension) producer.retry per retriable attempt

Example:

from klient.metrics import snapshot
print(snapshot())  # {'producer.tx.begin': 12, 'consumer.rebalance.assign': 3, ...}

Prometheus integration can wrap snapshot() periodically; for large-scale multi-process exporters use shared storage (Redis/memfd) or native client libs.

Relay Concurrency

ExactlyOnceRelay.arun(..., max_in_flight=N) allows up to N transactional batches in-flight concurrently (semaphore controlled). This improves throughput for high-latency commit scenarios while preserving batch atomicity. Tune batch_size + max_in_flight to balance memory and latency.

Isolation & Visibility

Use read_committed consumers when relaying from transactional producers to avoid forwarding aborted messages. The relay commits source offsets only after a successful target transaction commit.

Extending Resiliency

Planned enhancements: configurable circuit breaker (cooldown window), pluggable metrics sink, dynamic error code refresh. Contributions welcome—keep changes focused and tested.

• KafkaProducerFencedError (fencing / competing transactional id)

Consumer:

• KafkaConsumerError (base)
• KafkaConsumerRetriableError (safe to retry poll)
• KafkaConsumerFatalError (stop consumption)
• KafkaConsumerRebalanceError (issues in assignment callbacks)

Rebalance Callbacks

Pass on_assign, on_revoke, on_lost to KafkaConsumer constructor. Exceptions raised inside callbacks are wrapped in KafkaConsumerRebalanceError to surface issues without silent failure.

Retriable Produce

Use:

producer.produce_with_retry("topic", value=b"data", max_attempts=5, base_backoff=0.05)

Async:

await producer.aproduce_with_retry("topic", value=b"data")

Populate RETRIABLE_ERROR_CODES, FATAL_ERROR_CODES, FENCING_ERROR_CODES in producer.py for environment-specific tuning.

Exactly-Once Relay Helper

Module: klient.relay.ExactlyOnceRelay

from klient.consumer import KafkaConsumer, ConsumerConfig
from klient.producer import KafkaProducer, ProducerConfig
from klient.relay import ExactlyOnceRelay

cons = KafkaConsumer(ConsumerConfig(bootstrap_servers="localhost:9092", group_id="relay", isolation_level="read_committed"))
prod = KafkaProducer(ProducerConfig(bootstrap_servers="localhost:9092", transactional_id="relay-tx"))

def transform(msg):
  return msg  # identity or modify msg.value

relay = ExactlyOnceRelay(cons, prod, transform=transform, batch_size=25)
relay.run("input-topic", "output-topic")  # blocking
 

Async variant:

await relay.arun("input-topic", "output-topic")

Guarantee model: offsets are committed only after transaction success, reducing risk of duplicate downstream visibility while still relying on broker EOS guarantees.

Multi-Topic Subscription

Subscribe with more than one topic by passing a list:

consumer.subscribe(["orders", "payments", "audit-events"])  # list-only API

When using the CLI, you still specify a single topic per invocation; for multi-topic inspection create a small script using the library.

Observability Highlights

Implemented:

• Counters for producer.tx.begin|commit|abort, rebalance events, and shutdown lifecycle
• Transaction duration logging (TransactionResult.duration_ms in JSON logs)
• Structured JSON logs for transaction lifecycle, rebalance callbacks, stream shutdown events (shutdown_requested, stream_ended)
• Retry backoff with jitter and circuit breaker open/close events
• Async relay concurrency control (max_in_flight) for higher throughput

Shutdown JSON events (example):

{"event":"shutdown_requested","signal":2,"topic":"events"}
{"event":"stream_ended","reason":"signal","messages":42,"duration_s":3.417}

These are emitted directly to stdout; integrate with log pipeline by tailing the process output.

Graceful Shutdown

The stream consumer (consume stream) installs SIGINT/SIGTERM handlers. On first signal:

1. Counter consumer.shutdown.signal increments
2. A JSON line {"event":"shutdown_requested",...} is emitted
3. Loop enters a grace window (--grace-period, default 2s) allowing in-flight processing to finish
4. After grace, offsets commit (if auto-commit enabled) and the loop ends
5. Counter consumer.shutdown.complete increments and a final stream_ended JSON event prints

Set --grace-period 0 for immediate termination (still emits both events). Use longer periods for at-least-once downstream processing guarantees.

Continuous Transactional Relay (Library First)

While a CLI helper exists, production integrations typically import the wrappers directly for clearer control, richer transformation logic, and structured instrumentation. Below are canonical synchronous and asynchronous patterns.

Sync Relay with Transactions

from klient.consumer import KafkaConsumer, ConsumerConfig
from klient.producer import KafkaProducer, ProducerConfig

SOURCE = "input-topic"
TARGET = "output-topic"

consumer = KafkaConsumer(ConsumerConfig(
  bootstrap_servers="localhost:9092",
  group_id="relay-sync",
  topics=[SOURCE],
  isolation_level="read_committed",
  enable_auto_commit=False,  # commit only after successful target transaction
))

producer = KafkaProducer(ProducerConfig(
  bootstrap_servers="localhost:9092",
  transactional_id="relay-sync-tx"
))

def transform_value(value_bytes: bytes) -> bytes:
  """Example transform: append a marker; return bytes."""
  base = value_bytes.decode() if value_bytes else ""
  return f"{base}|processed".encode()

batch = []
BATCH_SIZE = 25

try:
  for msg in consumer.message_stream(timeout=0.5):
    batch.append(msg)
    if len(batch) >= BATCH_SIZE:
      producer.begin_transaction()
      for m in batch:
        transformed = transform_value(m.value)
        producer.produce(TARGET, key=m.key.decode() if m.key else None, value=transformed, flush=False)
      producer.commit_transaction()
      consumer.commit()  # commit source offsets only after transaction success
      batch.clear()
except KeyboardInterrupt:
  # Allow current batch to finish or abort.
  if producer.in_transaction:
    producer.abort_transaction()
finally:
  consumer.stop()
  producer.close()

Async Relay with Concurrency

import asyncio
from klient.consumer import KafkaConsumer, ConsumerConfig
from klient.producer import KafkaProducer, ProducerConfig

SOURCE = "input-topic"
TARGET = "output-topic"

consumer = KafkaConsumer(ConsumerConfig(
  bootstrap_servers="localhost:9092",
  group_id="relay-async",
  topics=[SOURCE],
  isolation_level="read_committed",
  enable_auto_commit=False,
))

producer = KafkaProducer(ProducerConfig(
  bootstrap_servers="localhost:9092",
  transactional_id="relay-async-tx"
))

stop_event = asyncio.Event()

def _sigint_handler():
  stop_event.set()

async def transform_bytes(b: bytes) -> bytes:
  # Illustrative async transform (could call external service)
  await asyncio.sleep(0)  # yield
  text = b.decode() if b else ""
  return f"{text}|async".encode()

async def relay_forever(batch_size: int = 50):
  buffer = []
  async for msg in consumer.amessage_stream(timeout=0.5):
    if stop_event.is_set():
      break
    buffer.append(msg)
    if len(buffer) >= batch_size:
      async with producer.atransaction():
        for m in buffer:
          transformed = await transform_bytes(m.value)
          await producer.aproduce(TARGET, key=m.key.decode() if m.key else None, value=transformed)
      consumer.commit()
      buffer.clear()
  # Flush residual messages (optional):
  if buffer:
    async with producer.atransaction():
      for m in buffer:
        transformed = await transform_bytes(m.value)
        await producer.aproduce(TARGET, key=m.key.decode() if m.key else None, value=transformed)
    consumer.commit()

async def main():
  loop = asyncio.get_running_loop()
  try:
    loop.add_signal_handler(getattr(__import__('signal'), 'SIGINT'), _sigint_handler)
  except (NotImplementedError, ValueError):
    pass
  await relay_forever()
  consumer.stop()
  producer.close()

asyncio.run(main())

Key points:

1. Source offsets are committed only after successful target transaction commit (exactly-once style).
2. Transform can be sync or async; ensure deterministic output for idempotence.
3. Graceful cancellation waits for current transaction scope to finish (avoid partial batch). Use a timeout wrapper if hard bounds are required.
4. For high throughput, shard by key/partition or use ExactlyOnceRelay with max_in_flight (async) for parallel in-flight transactions.
5. See examples/continuous_relay_async.py for a complete async variant with retry and error handling.

Continuous Transactional Relay (CLI)

A common pattern is to consume from one topic, transform each message, and produce to another topic atomically in transactional batches. The CLI provides a relay stream command for this:

python -m klient relay stream source-topic target-topic \
  --group relay-g1 \
  --transactional-id relay-tx-id \
  --batch-size 50 \
  --timeout 0.5 \
  --grace-period 2 \
  --transform 'value.upper()'

Behavior:

• Buffers up to --batch-size messages from source-topic under --group.
• Begins a transaction, applies --transform expression to each message value (value bound to decoded UTF-8 string), produces all outputs to target-topic.
• Commits the transaction; on success commits source offsets (exactly-once style handoff) and emits relay_batch_committed JSON event.
• On transactional failure the batch is aborted (relay_batch_aborted event) and offsets are NOT committed (messages will be retried on next pass).
• Graceful shutdown via SIGINT/SIGTERM triggers shutdown_requested + waits up to --grace-period seconds before final relay_stream_ended event.

Metrics involved:

• relay.batch.commit increments per committed batch.
• relay.messages.forwarded aggregates total messages successfully forwarded.
• consumer.shutdown.signal / consumer.shutdown.complete for lifecycle.

Library alternative for tighter control or custom transformation logic can use ExactlyOnceRelay directly or write an async loop with atransaction() contexts.

Future Enhancements

• Populate error code sets with actual KafkaError codes
• Backoff jitter & circuit breaker after repeated failures
• Configurable max in-flight transactions for high-throughput relays
• Pluggable metrics reporter interface

---

Troubleshooting

Symptom	Suggestion
No messages consumed	Check topic name, group id, and that messages are actually produced. Increase timeout.
Abort errors in transactions	Ensure broker supports transactions and transactional.id is stable per producer instance.
Config env not found	Verify filename or presence of default block in combined config.json.
Unexpected duplicates	Consider enabling idempotence (auto when transactional_id set).

---

License

MIT (or organization internal) — update as appropriate.