Redpanda

# Kafka 클라이언트로 Redpanda 연결 (100% 호환)
from kafka import KafkaProducer, KafkaConsumer
from kafka.admin import KafkaAdminClient, NewTopic
import json

REDPANDA_BROKERS = ['localhost:9092']

# Admin: 토픽 생성
admin_client = KafkaAdminClient(
    bootstrap_servers=REDPANDA_BROKERS,
    client_id='admin-client'
)

topic = NewTopic(
    name='user-events',
    num_partitions=6,
    replication_factor=3
)

try:
    admin_client.create_topics([topic])
    print("Topic created")
except Exception as e:
    print(f"Topic exists or error: {e}")

# Producer: 메시지 발행
producer = KafkaProducer(
    bootstrap_servers=REDPANDA_BROKERS,
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',  # 모든 복제본 확인
    retries=3,
    linger_ms=5  # 배치 대기 시간
)

# 이벤트 발행
for i in range(100):
    event = {
        'user_id': f'user_{i % 10}',
        'action': 'click',
        'timestamp': '2024-01-15T10:30:00Z',
        'page': f'/product/{i}'
    }
    future = producer.send('user-events', value=event, key=event['user_id'].encode())
    # 동기 전송: future.get(timeout=10)

producer.flush()
producer.close()

# Consumer: 메시지 소비
consumer = KafkaConsumer(
    'user-events',
    bootstrap_servers=REDPANDA_BROKERS,
    group_id='analytics-consumer',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    print(f"Partition: {message.partition}, Offset: {message.offset}")
    print(f"Key: {message.key}, Value: {message.value}")

    # 처리 후 수동 커밋 (enable_auto_commit=False 시)
    # consumer.commit()

# rpk CLI 사용 예시 (Redpanda 전용 도구)
# rpk topic create user-events --partitions 6 --replicas 3
# rpk topic produce user-events
# rpk topic consume user-events --group my-group
# rpk cluster info
# rpk topic describe user-events

# Docker로 Redpanda 실행
# docker run -d --name redpanda \
#   -p 9092:9092 -p 8081:8081 -p 8082:8082 \
#   docker.redpanda.com/redpandadata/redpanda:latest \
#   redpanda start --smp 1 --memory 1G --overprovisioned