Apache Kafka

# === Python: kafka-python 또는 confluent-kafka ===
from kafka import KafkaProducer, KafkaConsumer
from kafka.admin import KafkaAdminClient, NewTopic
import json

# 1. Producer 생성
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',  # 모든 복제본이 기록 확인 후 응답
    retries=3,
    batch_size=16384,  # 배치 크기 (bytes)
    linger_ms=10       # 배치 대기 시간 (ms)
)

# 메시지 발행
def send_order(order):
    future = producer.send(
        topic='orders',
        key=order['order_id'],
        value=order,
        headers=[('source', b'order-service')]
    )
    # 동기 전송 (응답 대기)
    result = future.get(timeout=10)
    print(f"Sent to partition {result.partition}, offset {result.offset}")

send_order({
    'order_id': 'order-123',
    'user_id': 'user-456',
    'total': 59900,
    'items': [{'sku': 'PROD-001', 'qty': 2}]
})

producer.flush()  # 모든 메시지 전송 완료 대기

# 2. Consumer 생성
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processor',
    auto_offset_reset='earliest',  # 처음부터 읽기 (latest: 최신부터)
    enable_auto_commit=True,
    auto_commit_interval_ms=5000,
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

# 메시지 소비
def process_orders():
    for message in consumer:
        order = message.value
        print(f"[P{message.partition}:O{message.offset}] Order: {order['order_id']}")
        # 비즈니스 로직 처리
        # consumer.commit()  # 수동 커밋 시

# 3. Admin: Topic 생성
admin = KafkaAdminClient(bootstrap_servers=['localhost:9092'])

topic = NewTopic(
    name='orders',
    num_partitions=6,
    replication_factor=3,
    topic_configs={
        'retention.ms': '604800000',  # 7일
        'cleanup.policy': 'delete'
    }
)
admin.create_topics([topic])

# === docker-compose.yml: Kafka + Zookeeper 로컬 환경 ===
version: "3.9"

services:
  # ZooKeeper (Kafka 메타데이터 관리) - KRaft 모드로 대체 가능
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  # Kafka Broker
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      # 로그 보존 설정
      KAFKA_LOG_RETENTION_HOURS: 168  # 7일
      KAFKA_LOG_SEGMENT_BYTES: 1073741824  # 1GB
      KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 300000

  # Kafka UI (웹 관리 도구)
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    depends_on:
      - kafka
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181

# === CLI 명령어 ===
# Topic 생성
# docker exec kafka kafka-topics --create --topic orders --partitions 6 --replication-factor 1 --bootstrap-server localhost:9092

# Topic 목록 확인
# docker exec kafka kafka-topics --list --bootstrap-server localhost:9092

# Consumer Group 상태 확인
# docker exec kafka kafka-consumer-groups --describe --group order-processor --bootstrap-server localhost:9092

# 메시지 확인 (콘솔 컨슈머)
# docker exec kafka kafka-console-consumer --topic orders --from-beginning --bootstrap-server localhost:9092