ApacheKafkaによるリアルタイムデータストリーミング:実践ガイド

# Producer: route by user_id to guarantee per-user ordering
producer.produce(
    topic="user-events",
    key=str(user_id).encode(),   # same key → same partition
    value=json.dumps(event).encode()
)

from confluent_kafka import Producer
import json
import time

conf = {
    "bootstrap.servers": "kafka1:9092,kafka2:9092,kafka3:9092",
    "acks": "all",                    # wait for all in-sync replicas to acknowledge
    "retries": 5,
    "retry.backoff.ms": 200,
    "compression.type": "lz4",        # compress batches — massive throughput boost
    "linger.ms": 5,                    # batch for up to 5ms before sending
    "batch.size": 65536,               # 64 KB batch size
    "enable.idempotence": True,        # exactly-once producer semantics
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed for {msg.key()}: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

# Produce a batch of events
for i in range(1000):
    event = {"user_id": i % 100, "action": "page_view", "ts": time.time()}
    producer.produce(
        topic="user-events",
        key=str(event["user_id"]),
        value=json.dumps(event),
        callback=delivery_report
    )
    producer.poll(0)    # trigger callbacks for previously sent messages

producer.flush()        # wait for all in-flight messages to be delivered

from confluent_kafka import Consumer, KafkaException
import json

conf = {
    "bootstrap.servers": "kafka1:9092,kafka2:9092,kafka3:9092",
    "group.id": "user-event-processor-v1",
    "auto.offset.reset": "earliest",     # start from the beginning if no committed offset
    "enable.auto.commit": False,          # manual commit for exactly-once processing
    "max.poll.interval.ms": 300000,       # 5 min max processing time per batch
    "session.timeout.ms": 30000,
}

consumer = Consumer(conf)
consumer.subscribe(["user-events"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)

        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())

        event = json.loads(msg.value().decode("utf-8"))

        # Process the event
        process_event(event)

        # Commit offset only AFTER successful processing
        consumer.commit(asynchronous=False)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

kafka-topics.sh --alter --topic user-events --partitions 24 \
  --bootstrap-server kafka1:9092

import asyncio
from asyncio import Queue

async def consume(consumer: Consumer, queue: Queue):
    while True:
        msg = consumer.poll(timeout=0.1)
        if msg and not msg.error():
            await queue.put(msg)       # blocks if queue is full — natural backpressure
            
async def process(queue: Queue):
    while True:
        msg = await queue.get()
        event = json.loads(msg.value())
        await process_event_async(event)
        queue.task_done()

# Bounded queue: max 500 in-flight events
queue = Queue(maxsize=500)
await asyncio.gather(consume(consumer, queue), process(queue))

def process_with_dlq(msg):
    for attempt in range(3):
        try:
            process_event(json.loads(msg.value()))
            return
        except Exception as e:
            if attempt == 2:
                # Send to DLQ with original headers + error metadata
                producer.produce("user-events.DLQ", value=msg.value(),
                                 headers={"error": str(e), "original-topic": "user-events"})

kafka-configs.sh --bootstrap-server kafka1:9092 \
  --entity-type topics \
  --entity-name user-profiles \
  --alter \
  --add-config "cleanup.policy=compact,min.cleanable.dirty.ratio=0.1,delete.retention.ms=86400000"

from confluent_kafka import Producer

conf = {
    "bootstrap.servers": "kafka1:9092",
    "transactional.id": "my-producer-instance-1",   # unique per producer instance
    "enable.idempotence": True,
}

producer = Producer(conf)
producer.init_transactions()

try:
    producer.begin_transaction()
    
    # Read from consumer, process, write result — all in one transaction
    for msg in batch:
        result = transform(json.loads(msg.value()))
        producer.produce("processed-events", value=json.dumps(result))
    
    # Commit consumer offsets and producer messages atomically
    producer.send_offsets_to_transaction(consumer_offsets, consumer.consumer_group_metadata())
    producer.commit_transaction()
    
except Exception:
    producer.abort_transaction()
    raise

# 1. Add new broker to cluster — it joins automatically
# 2. Reassign partitions to include new broker
kafka-reassign-partitions.sh \
  --bootstrap-server kafka1:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "1,2,3,4" \          # include new broker 4
  --generate

# 3. Execute the reassignment plan
kafka-reassign-partitions.sh \
  --bootstrap-server kafka1:9092 \
  --reassignment-json-file reassignment.json \
  --execute \
  --throttle 50000000                 # 50 MB/s replication throttle — don't saturate network