CDC

// Kafka Consumer로 CDC 이벤트 처리 (Java)
import org.apache.kafka.clients.consumer.*;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class CDCEventConsumer {
    private final ObjectMapper mapper = new ObjectMapper();

    public void consumeCDCEvents() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka:9092");
        props.put("group.id", "order-sync-service");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("cdc.orders"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    processCDCEvent(record.value());
                }

                consumer.commitSync();  // 처리 완료 후 커밋
            }
        }
    }

    private void processCDCEvent(String eventJson) throws Exception {
        JsonNode event = mapper.readTree(eventJson);
        String operation = event.get("op").asText();

        switch (operation) {
            case "c" -> handleCreate(event.get("after"));
            case "u" -> handleUpdate(event.get("before"), event.get("after"));
            case "d" -> handleDelete(event.get("before"));
            case "r" -> handleSnapshot(event.get("after"));
        }
    }

    private void handleUpdate(JsonNode before, JsonNode after) {
        Long orderId = after.get("id").asLong();
        String oldStatus = before.get("status").asText();
        String newStatus = after.get("status").asText();

        if (!oldStatus.equals(newStatus)) {
            System.out.println("Order " + orderId + ": " + oldStatus + " → " + newStatus);
            // 읽기 모델 업데이트, 알림 발송 등
            updateReadModel(orderId, after);
            sendNotification(orderId, newStatus);
        }
    }
}

# Python으로 CDC 이벤트 처리
from kafka import KafkaConsumer
from dataclasses import dataclass
from typing import Optional, Dict, Any
import json

@dataclass
class CDCEvent:
    operation: str  # c, u, d, r
    before: Optional[Dict[str, Any]]
    after: Optional[Dict[str, Any]]
    source: Dict[str, Any]
    timestamp: int

    @classmethod
    def from_json(cls, data: dict) -> "CDCEvent":
        return cls(
            operation=data.get("op"),
            before=data.get("before"),
            after=data.get("after"),
            source=data.get("source", {}),
            timestamp=data.get("ts_ms", 0)
        )


class OrderCDCProcessor:
    def __init__(self, kafka_servers: str):
        self.consumer = KafkaConsumer(
            "cdc.orders",
            bootstrap_servers=kafka_servers,
            group_id="order-analytics",
            value_deserializer=lambda m: json.loads(m.decode("utf-8")),
            auto_offset_reset="earliest",
            enable_auto_commit=False
        )

    def process_events(self):
        """CDC 이벤트 처리 루프"""
        for message in self.consumer:
            try:
                event = CDCEvent.from_json(message.value)
                self.handle_event(event)
                self.consumer.commit()
            except Exception as e:
                print(f"Error processing event: {e}")
                # Dead Letter Queue로 전송

    def handle_event(self, event: CDCEvent):
        """이벤트 타입별 처리"""
        handlers = {
            "c": self.on_create,
            "u": self.on_update,
            "d": self.on_delete,
            "r": self.on_snapshot
        }

        handler = handlers.get(event.operation)
        if handler:
            handler(event)

    def on_create(self, event: CDCEvent):
        """새 주문 생성"""
        order = event.after
        print(f"✅ New order: #{order['id']} - {order['total']}원")
        # 분석 시스템에 전송
        self.send_to_analytics("order_created", order)

    def on_update(self, event: CDCEvent):
        """주문 상태 변경"""
        before, after = event.before, event.after

        # 변경된 필드 감지
        changes = {
            k: (before.get(k), after.get(k))
            for k in after.keys()
            if before.get(k) != after.get(k)
        }

        if "status" in changes:
            old, new = changes["status"]
            print(f"📦 Order #{after['id']}: {old} → {new}")

            if new == "shipped":
                self.notify_shipping(after)
            elif new == "delivered":
                self.update_inventory(after)

    def on_delete(self, event: CDCEvent):
        """주문 삭제 (Soft delete 권장)"""
        order = event.before
        print(f"🗑️ Order deleted: #{order['id']}")
        # 인덱스에서 제거
        self.remove_from_search_index(order['id'])

    def send_to_analytics(self, event_type: str, data: dict):
        """분석 시스템으로 전송"""
        # ClickHouse, BigQuery 등으로 적재
        pass


if __name__ == "__main__":
    processor = OrderCDCProcessor("kafka:9092")
    processor.process_events()