from confluent_kafka import Producer, KafkaError
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
import json
import uuid
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
@dataclass
class OrderCreatedEvent:
event_id: str
order_id: str
user_id: str
total_cents: int
items: list[dict]
created_at: str
schema_version: str = "1.0"
class KafkaEventProducer:
def __init__(self, bootstrap_servers: str, schema_registry_url: str = None):
self._producer = Producer({
"bootstrap.servers": bootstrap_servers,
"acks": "all",
"enable.idempotence": True,
"max.in.flight.requests.per.connection": 5,
"retries": 2147483647,
"delivery.timeout.ms": 30000,
"compression.type": "snappy",
"batch.size": 65536,
"linger.ms": 5,
})
self._dlq_topic = "dlq-events"
def publish(
self,
topic: str,
event: dict,
key: str = None,
headers: dict = None
) -> None:
"""Publish event with delivery guarantee."""
envelope = {
**event,
"event_id": event.get("event_id") or str(uuid.uuid4()),
"published_at": datetime.now(timezone.utc).isoformat(),
}
kafka_headers = []
if headers:
for k, v in headers.items():
kafka_headers.append((k, str(v).encode()))
def delivery_callback(err, msg):
if err:
logger.error(
f"Message delivery failed: {err}",
extra={"topic": topic, "key": key}
)
self._send_to_dlq(topic, key, envelope, str(err))
else:
logger.debug(
f"Message delivered to {msg.topic()} [{msg.partition()}] @{msg.offset()}"
)
self._producer.produce(
topic=topic,
key=key.encode() if key else None,
value=json.dumps(envelope).encode(),
headers=kafka_headers,
on_delivery=delivery_callback
)
def flush(self, timeout: float = 10.0):
remaining = self._producer.flush(timeout=timeout)
if remaining > 0:
raise TimeoutError(f"{remaining} messages not delivered after {timeout}s")
def _send_to_dlq(self, original_topic: str, key: str, event: dict, error: str):
dlq_event = {
"original_topic": original_topic,
"original_key": key,
"original_event": event,
"error": error,
"failed_at": datetime.now(timezone.utc).isoformat(),
}
self._producer.produce(
topic=self._dlq_topic,
key=key.encode() if key else None,
value=json.dumps(dlq_event).encode()
)
from confluent_kafka import Consumer, KafkaError, TopicPartition
import signal
import threading
class KafkaEventConsumer:
def __init__(
self,
bootstrap_servers: str,
group_id: str,
topics: list[str],
max_poll_interval_ms: int = 300000
):
self._consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": group_id,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"max.poll.interval.ms": max_poll_interval_ms,
"session.timeout.ms": 30000,
"heartbeat.interval.ms": 3000,
"fetch.min.bytes": 1,
"fetch.max.wait.ms": 500,
})
self._consumer.subscribe(topics)
self._running = True
def consume(self, handler, batch_size: int = 10, timeout_ms: int = 1000):
"""Process messages with at-least-once delivery semantics."""
while self._running:
messages = self._consumer.consume(
num_messages=batch_size,
timeout=timeout_ms / 1000
)
if not messages:
continue
failed_offsets = []
for msg in messages:
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
logger.error(f"Consumer error: {msg.error()}")
continue
try:
event = json.loads(msg.value())
handler(event, msg.headers() or [])
except Exception as e:
logger.error(
f"Failed to process message: {e}",
extra={
"topic": msg.topic(),
"partition": msg.partition(),
"offset": msg.offset(),
}
)
failed_offsets.append((msg.topic(), msg.partition(), msg.offset()))
continue
if not failed_offsets:
self._consumer.commit(asynchronous=False)
def stop(self):
self._running = False
self._consumer.close()
TOPIC_CONFIG = {
"orders.created": {
"partitions": 12,
"replication_factor": 3,
"retention.ms": 7 * 24 * 60 * 60 * 1000,
"cleanup.policy": "delete",
"compression.type": "snappy",
"min.insync.replicas": 2,
},
"orders.status-updates": {
"partitions": 6,
"replication_factor": 3,
"retention.ms": 3 * 24 * 60 * 60 * 1000,
},
"dlq-events": {
"partitions": 3,
"replication_factor": 3,
"retention.ms": 30 * 24 * 60 * 60 * 1000,
}
}
import pika
import json
import time
from typing import Callable
class RabbitMQSetup:
"""Exchange + queue topology for order processing."""
def __init__(self, connection_url: str):
self.connection = pika.BlockingConnection(
pika.URLParameters(connection_url)
)
self.channel = self.connection.channel()
self.channel.basic_qos(prefetch_count=1)
def setup_topology(self):
"""Declare exchanges, queues, and bindings."""
self.channel.exchange_declare(
exchange="dlx",
exchange_type="direct",
durable=True
)
self.channel.queue_declare(
queue="dead-letter",
durable=True,
arguments={"x-queue-type": "quorum"}
)
self.channel.queue_bind(
exchange="dlx",
queue="dead-letter",
routing_key="#"
)
self.channel.exchange_declare(
exchange="orders",
exchange_type="topic",
durable=True
)
self.channel.queue_declare(
queue="order-processor",
durable=True,
arguments={
"x-queue-type": "quorum",
"x-dead-letter-exchange": "dlx",
"x-dead-letter-routing-key": "order-processor",
"x-message-ttl": 600000,
}
)
self.channel.queue_bind(
exchange="orders",
queue="order-processor",
routing_key="order.created"
)
self.channel.queue_declare(
queue="order-notifications",
durable=True,
arguments={
"x-queue-type": "quorum",
"x-dead-letter-exchange": "dlx",
}
)
self.channel.queue_bind(
exchange="orders",
queue="order-notifications",
routing_key="order.*"
)
class RabbitMQProducer:
def __init__(self, channel):
self.channel = channel
def publish(self, routing_key: str, event: dict, priority: int = 0):
self.channel.basic_publish(
exchange="orders",
routing_key=routing_key,
body=json.dumps(event).encode(),
properties=pika.BasicProperties(
delivery_mode=pika.DeliveryMode.Persistent,
content_type="application/json",
message_id=event.get("event_id"),
timestamp=int(time.time()),
priority=priority,
)
)
class RabbitMQConsumer:
def __init__(self, channel, queue: str, max_retries: int = 3):
self.channel = channel
self.queue = queue
self.max_retries = max_retries
def consume(self, handler: Callable):
def callback(ch, method, properties, body):
retry_count = int(
(properties.headers or {}).get("x-retry-count", 0)
)
try:
event = json.loads(body)
handler(event)
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
logger.error(f"Processing failed (attempt {retry_count + 1}): {e}")
if retry_count < self.max_retries:
headers = properties.headers or {}
headers["x-retry-count"] = retry_count + 1
delay_ms = 1000 * (2 ** retry_count)
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
self.channel.basic_publish(
exchange="orders",
routing_key=method.routing_key,
body=body,
properties=pika.BasicProperties(
headers={**headers, "x-delay": delay_ms},
delivery_mode=pika.DeliveryMode.Persistent,
)
)
else:
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
self.channel.basic_consume(
queue=self.queue,
on_message_callback=callback
)
self.channel.start_consuming()
import boto3
import json
import uuid
from datetime import datetime, timezone
class SQSConsumer:
def __init__(self, queue_url: str, max_messages: int = 10):
self.sqs = boto3.client("sqs")
self.queue_url = queue_url
self.max_messages = max_messages
def process(self, handler: callable, delete_on_success: bool = True):
while True:
response = self.sqs.receive_message(
QueueUrl=self.queue_url,
MaxNumberOfMessages=self.max_messages,
WaitTimeSeconds=20,
MessageAttributeNames=["All"],
AttributeNames=["ApproximateReceiveCount"]
)
for message in response.get("Messages", []):
receive_count = int(
message.get("Attributes", {}).get("ApproximateReceiveCount", 1)
)
try:
body = json.loads(message["Body"])
if "Message" in body:
event = json.loads(body["Message"])
else:
event = body
handler(event)
if delete_on_success:
self.sqs.delete_message(
QueueUrl=self.queue_url,
ReceiptHandle=message["ReceiptHandle"]
)
except Exception as e:
logger.error(f"Processing failed (receive #{receive_count}): {e}")
def create_queue_infrastructure():
sqs = boto3.client("sqs")
dlq = sqs.create_queue(
QueueName="orders-dlq.fifo",
Attributes={
"FifoQueue": "true",
"MessageRetentionPeriod": str(14 * 24 * 3600),
}
)
dlq_arn = sqs.get_queue_attributes(
QueueUrl=dlq["QueueUrl"],
AttributeNames=["QueueArn"]
)["Attributes"]["QueueArn"]
main_queue = sqs.create_queue(
QueueName="orders.fifo",
Attributes={
"FifoQueue": "true",
"ContentBasedDeduplication": "true",
"VisibilityTimeout": "300",
"MessageRetentionPeriod": str(4 * 24 * 3600),
"RedrivePolicy": json.dumps({
"deadLetterTargetArn": dlq_arn,
"maxReceiveCount": "3"
})
}
)
import redis
from functools import wraps
class IdempotencyStore:
def __init__(self, redis_client, ttl: int = 86400):
self.redis = redis_client
self.ttl = ttl
def is_processed(self, event_id: str) -> bool:
return bool(self.redis.exists(f"processed:{event_id}"))
def mark_processed(self, event_id: str):
self.redis.setex(f"processed:{event_id}", self.ttl, "1")
def idempotent(idempotency_store: IdempotencyStore):
"""Decorator: skip duplicate events."""
def decorator(func):
@wraps(func)
def wrapper(event: dict, *args, **kwargs):
event_id = event.get("event_id")
if not event_id:
logger.warning("Event missing event_id — cannot ensure idempotency")
return func(event, *args, **kwargs)
if idempotency_store.is_processed(event_id):
logger.info(f"Skipping duplicate event {event_id}")
return
result = func(event, *args, **kwargs)
idempotency_store.mark_processed(event_id)
return result
return wrapper
return decorator
store = IdempotencyStore(redis_client)
@idempotent(store)
def handle_order_created(event: dict):
order = create_order_in_db(event)
send_confirmation_email(order)
update_inventory(order)
