Kafka #

# Library Python untuk Kafka
pip install kafka-python

# Alternatif: confluent-kafka (binding librdkafka, lebih performan untuk produksi)
pip install confluent-kafka

# docker-compose.yml untuk Kafka dengan KRaft (tanpa Zookeeper)
# Simpan sebagai docker-compose.yml dan jalankan: docker compose up -d

version: "3"
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_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

import json
import os
from kafka import KafkaProducer
from kafka.errors import KafkaError

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# ANTI-PATTERN: kirim bytes mentah tanpa serialisasi
producer = KafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS)
producer.send("orders", b"{'order_id': 1}")  # ✗ -- bukan JSON valid, sulit di-consume

# BENAR: konfigurasi serializer JSON
producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
    key_serializer=lambda k: k.encode("utf-8") if k else None,
    acks="all",             # tunggu konfirmasi dari semua replika (paling aman)
    retries=3,              # coba ulang jika gagal
    retry_backoff_ms=300,   # jeda antar retry
    request_timeout_ms=30000,
    compression_type="gzip" # kompres pesan untuk efisiensi bandwidth
)

def kirim_order(producer: KafkaProducer, order: dict) -> None:
    topic   = "orders"
    # Key menentukan partition -- pesan dengan key sama selalu ke partition yang sama
    # Gunakan ID entitas sebagai key untuk menjaga urutan per entitas
    key     = str(order.get("order_id"))

    future = producer.send(topic, key=key, value=order)

    try:
        record_metadata = future.get(timeout=10)  # tunggu konfirmasi
        print(
            f"Pesan terkirim → topic: {record_metadata.topic}, "
            f"partition: {record_metadata.partition}, "
            f"offset: {record_metadata.offset}"
        )
    except KafkaError as e:
        print(f"Gagal mengirim pesan: {e}")

# Penggunaan
order = {
    "order_id":   1001,
    "pengguna_id": 42,
    "produk":     "Laptop Gaming ASUS",
    "total":      18500000,
    "status":     "pending",
    "timestamp":  "2024-03-15T10:30:00Z"
}

kirim_order(producer, order)
producer.flush()  # pastikan semua pesan terkirim sebelum exit

def on_send_success(record_metadata):
    print(
        f"✓ Terkirim → {record_metadata.topic}:"
        f"[{record_metadata.partition}]@{record_metadata.offset}"
    )

def on_send_error(exc):
    print(f"✗ Gagal mengirim: {exc}")

def kirim_event_async(producer: KafkaProducer, topic: str, key: str, event: dict) -> None:
    producer.send(topic, key=key, value=event) \
            .add_callback(on_send_success)     \
            .add_errback(on_send_error)

# Kirim banyak event secara asinkron
events = [
    {"event": "user_login",    "user_id": 1, "ip": "192.168.1.1"},
    {"event": "product_view",  "user_id": 1, "product_id": 101},
    {"event": "add_to_cart",   "user_id": 1, "product_id": 101},
]

for event in events:
    kirim_event_async(producer, "user-events", str(event["user_id"]), event)

producer.flush()  # tunggu semua callback selesai
producer.close()

from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import signal

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# ANTI-PATTERN: enable_auto_commit=True tanpa memahami risikonya
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    enable_auto_commit=True,   # ✗ -- commit otomatis sebelum pemrosesan selesai
    group_id="order-service"   #    bisa menyebabkan pesan hilang jika crash saat proses
)

# BENAR: manual commit setelah pemrosesan berhasil
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    group_id="order-service",
    auto_offset_reset="earliest",   # mulai dari awal jika group baru / offset tidak ada
                                    # "latest" = hanya baca pesan baru
    enable_auto_commit=False,       # commit manual setelah pemrosesan berhasil
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    key_deserializer=lambda k: k.decode("utf-8") if k else None,
    session_timeout_ms=30000,
    heartbeat_interval_ms=10000,
    max_poll_records=50             # maksimum pesan per poll
)

import signal
import threading

def proses_order(order: dict) -> None:
    """Logika bisnis pemrosesan order."""
    print(f"Memproses order #{order['order_id']} — total Rp{order['total']:,.0f}")
    # ... simpan ke database, kirim notifikasi, dll.

def jalankan_consumer():
    berjalan = True

    def handle_shutdown(signum, frame):
        nonlocal berjalan
        print("Menerima sinyal shutdown, menghentikan consumer...")
        berjalan = False

    signal.signal(signal.SIGINT,  handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    consumer = KafkaConsumer(
        "orders",
        bootstrap_servers=BOOTSTRAP_SERVERS,
        group_id="order-service",
        auto_offset_reset="earliest",
        enable_auto_commit=False,
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        key_deserializer=lambda k: k.decode("utf-8") if k else None,
    )

    print("Consumer aktif, menunggu pesan...")
    try:
        while berjalan:
            # poll() dengan timeout agar bisa cek flag berjalan secara periodik
            records = consumer.poll(timeout_ms=1000, max_records=50)

            for topic_partition, messages in records.items():
                for msg in messages:
                    try:
                        proses_order(msg.value)
                        # Commit hanya setelah pemrosesan berhasil
                        consumer.commit()
                    except Exception as e:
                        print(f"Error memproses pesan offset {msg.offset}: {e}")
                        # Jangan commit -- pesan akan dibaca ulang setelah restart
                        # Tambahkan ke dead letter queue jika perlu

    finally:
        consumer.close()
        print("Consumer ditutup dengan bersih.")

jalankan_consumer()

# Diagram consumer group:
#
# Topic "orders" dengan 3 partition:
#
# Partition 0 ──────► Consumer A  ┐
# Partition 1 ──────► Consumer B  ├── Group: "order-service"
# Partition 2 ──────► Consumer C  ┘
#
# Jika Consumer B mati:
# Partition 0 ──────► Consumer A  ┐
# Partition 1 ──────► Consumer A  ├── Rebalance otomatis
# Partition 2 ──────► Consumer C  ┘

# Subscribe ke beberapa topic sekaligus
consumer.subscribe(["orders", "payments", "notifications"])

# Atau assign ke partition spesifik secara manual (tanpa rebalance)
from kafka import TopicPartition

consumer.assign([
    TopicPartition("orders", 0),
    TopicPartition("orders", 1),
])

# Cek assignment saat ini
print("Assigned partitions:", consumer.assignment())

# Cek lag (berapa pesan yang belum diproses)
for tp in consumer.assignment():
    committed = consumer.committed(tp)
    end       = consumer.end_offsets([tp])[tp]
    lag       = end - (committed.offset if committed else 0)
    print(f"{tp.topic}[{tp.partition}]: lag={lag}")

import json
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class OrderEvent:
    order_id:   int
    user_id:    int
    total:      float
    status:     str
    created_at: str = None

    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.utcnow().isoformat() + "Z"

# Producer dengan dataclass
producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(asdict(v), ensure_ascii=False).encode("utf-8"),
    key_serializer=lambda k: str(k).encode("utf-8") if k else None,
    acks="all"
)

event = OrderEvent(order_id=1001, user_id=42, total=18500000.0, status="created")
producer.send("order-events", key=event.order_id, value=event)
producer.flush()

# Consumer yang me-reconstruct dataclass
consumer = KafkaConsumer(
    "order-events",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    group_id="order-processor",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    value_deserializer=lambda v: OrderEvent(**json.loads(v.decode("utf-8")))
)

for msg in consumer:
    order: OrderEvent = msg.value
    print(f"Order #{order.order_id}: Rp{order.total:,.0f} — {order.status}")
    consumer.commit()

from kafka.admin import KafkaAdminClient, NewTopic
from kafka.errors import TopicAlreadyExistsError

def buat_topic(
    nama:         str,
    num_partitions: int = 3,
    replication_factor: int = 1
) -> None:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)

    topic = NewTopic(
        name=nama,
        num_partitions=num_partitions,
        replication_factor=replication_factor,
        topic_configs={
            "retention.ms":     str(7 * 24 * 60 * 60 * 1000),  # simpan 7 hari
            "cleanup.policy":   "delete",
            "compression.type": "gzip"
        }
    )

    try:
        admin.create_topics([topic])
        print(f"Topic '{nama}' berhasil dibuat dengan {num_partitions} partition.")
    except TopicAlreadyExistsError:
        print(f"Topic '{nama}' sudah ada.")
    finally:
        admin.close()

def list_topics() -> list[str]:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)
    topics = admin.list_topics()
    admin.close()
    return [t for t in topics if not t.startswith("__")]  # filter internal topics

def hapus_topic(nama: str) -> None:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)
    admin.delete_topics([nama])
    admin.close()
    print(f"Topic '{nama}' dihapus.")

# Buat topic untuk aplikasi
buat_topic("orders",       num_partitions=3)
buat_topic("order-events", num_partitions=3)
buat_topic("user-events",  num_partitions=6)
print("Topics:", list_topics())

from kafka import KafkaProducer, KafkaConsumer
import json

DLQ_TOPIC = "orders-dlq"   # Dead Letter Queue -- pesan yang gagal diproses

def proses_dengan_retry(
    consumer: KafkaConsumer,
    dlq_producer: KafkaProducer,
    maks_retry: int = 3
) -> None:

    for msg in consumer:
        retry_count = 0
        berhasil    = False

        while retry_count <= maks_retry and not berhasil:
            try:
                # Proses pesan
                proses_order(msg.value)
                consumer.commit()
                berhasil = True

            except Exception as e:
                retry_count += 1
                print(f"Retry {retry_count}/{maks_retry} untuk offset {msg.offset}: {e}")

                if retry_count > maks_retry:
                    # Kirim ke Dead Letter Queue
                    dlq_payload = {
                        "original_topic":     msg.topic,
                        "original_partition": msg.partition,
                        "original_offset":    msg.offset,
                        "original_key":       msg.key,
                        "original_value":     msg.value,
                        "error":              str(e),
                        "failed_at":          datetime.utcnow().isoformat()
                    }
                    dlq_producer.send(DLQ_TOPIC, value=dlq_payload)
                    dlq_producer.flush()
                    consumer.commit()  # commit agar tidak loop terus
                    print(f"Pesan dikirim ke DLQ: {DLQ_TOPIC}")