kafka

# docker-compose.yml — Kafka with KRaft (no ZooKeeper)
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      KAFKA_CFG_NODE_ID: 1
      KAFKA_CFG_PROCESS_ROLES: broker,controller
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER

// producer.ts — Send events to Kafka
import { Kafka, Partitioners } from 'kafkajs'

const kafka = new Kafka({ brokers: ['localhost:9092'] })
const producer = kafka.producer({ createPartitioner: Partitioners.DefaultPartitioner })

await producer.connect()

// Send single event
await producer.send({
  topic: 'orders',
  messages: [
    {
      key: 'order-123',                    // partition key (orders for same user go to same partition)
      value: JSON.stringify({
        orderId: 'order-123',
        userId: 'user-456',
        items: [{ sku: 'WIDGET-1', quantity: 2, price: 29.99 }],
        total: 59.98,
        createdAt: new Date().toISOString(),
      }),
    },
  ],
})

// Batch send
await producer.sendBatch({
  topicMessages: [
    { topic: 'orders', messages: events.map(e => ({ key: e.id, value: JSON.stringify(e) })) },
  ],
})

await producer.disconnect()

// consumer.ts — Process events from Kafka
const consumer = kafka.consumer({ groupId: 'order-service' })

await consumer.connect()
await consumer.subscribe({ topic: 'orders', fromBeginning: false })

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const order = JSON.parse(message.value.toString())
    console.log(`Processing order ${order.orderId} from partition ${partition}`)

    // Process the order (idempotently — messages can be redelivered)
    await processOrder(order)
  },
})

// Graceful shutdown
process.on('SIGTERM', async () => {
  await consumer.disconnect()
})

# consumer.py — Kafka consumer with confluent-kafka
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'analytics-service',
    'auto.offset.reset': 'earliest',
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Error: {msg.error()}")
        continue

    order = json.loads(msg.value().decode('utf-8'))
    print(f"Processing: {order['orderId']}")