kafka-platform/
├── brokers/
│   └── server.properties                # 브로커 공통 설정 참고본 (Hard 설정 주석 포함)
│
├── topics/                              # 토픽 선언 YAML — 토픽 추가 시 이 파일만 추가
│   ├── order/
│   │   ├── order.created.v1.yaml
│   │   └── order.cancelled.v1.yaml
│   └── log/
│       └── system.log.v1.yaml           # Connect DB sink 대상 로그 토픽
│
├── scripts/
│   └── create-topics.sh                 # topics/ YAML을 읽어 토픽 자동 생성
│
├── connectors/
│   └── db-sink/
│       ├── system-log-sink.json         # JDBC Sink Connector 설정
│       ├── system_log.ddl.sql           # 테이블 DDL
│       └── README.md                    # bulk insert 동작 원리, 환경변수 설명
│
└── test/
    ├── docker-compose.yml               # 로컬 개발용 (Kafka + Kafka UI + Redis + Nexus)
    ├── .env.dev                         # 로컬 개발 환경변수 (커밋)
    └── .env.qa                          # QA 환경변수 (커밋, 시크릿 제외)

name: prd.order.created.v1
partitions: 6
replication-factor: 3
configs:
  retention.ms: 604800000       # 7일
  min.insync.replicas: "2"      # Hard 설정
  cleanup.policy: delete

# 로컬 개발
docker compose --env-file .env.dev up -d

{
  "name": "system-log-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "prd.log.system.v1",
    "connection.url": "${env:DB_URL}",
    "connection.user": "${env:DB_USER}",
    "connection.password": "${env:DB_PASSWORD}",
    "insert.mode": "insert",
    "auto.create": "false",
    "auto.evolve": "false",
    "table.name.format": "system_log",
    "pk.mode": "none",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "batch.size": "10",
    "consumer.override.max.poll.records": "10",
    "consumer.override.fetch.min.bytes": "1024",
    "consumer.override.fetch.max.wait.ms": "1000",
    "transforms": "rename,drop",
    "transforms.rename.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.rename.renames": "eventId:event_id,aggregateId:aggregate_id,serviceId:service_id,occurredAt:occurred_at",
    "transforms.drop.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.drop.exclude": "context"
  }
}

kafka-common-lib/
├── build.gradle
│
└── src/main/java/.../kafka/
    ├── config/
    │   ├── KafkaProducerConfig.java     # Hard 설정 강제 (acks=all, idempotence)
    │   └── KafkaConsumerConfig.java     # isolation.level, Soft 설정 기본값
    │
    ├── events/                          # 이벤트 POJO (도메인별 패키지)
    │   ├── order/
    │   │   ├── OrderCreatedEvent.java
    │   │   └── OrderCancelledEvent.java
    │   └── log/
    │       └── SystemLogEvent.java      # Connect DB sink 대상 로그 이벤트
    │
    ├── idempotency/                     # annotation 기반 Redis 멱등성
    │   ├── IdempotentConsumer.java      # @annotation 정의
    │   ├── IdempotencyAspect.java       # AOP — annotation 감지·Redis 체크·skip
    │   └── IdempotencyRedisStore.java   # Redis SET NX + TTL
    │
    ├── error/
    │   └── DltPublisher.java            # Dead Letter Topic 발행
    │
    └── serde/
        └── JsonEventSerializer.java     # ObjectMapper 설정 (날짜 포맷 등)

public record OrderCreatedEvent(
    String eventId,          // UUID — 멱등성 키
    String aggregateId,      // orderId
    String customerId,
    List<OrderItem> items,
    BigDecimal totalAmount,
    Instant occurredAt
) {}

public record SystemLogEvent(
    String eventId,
    String serviceId,                  // 시스템 식별자 (예: "order-service")
    String level,                      // INFO / WARN / ERROR
    String message,
    Map<String, String> context,       // 추가 컨텍스트 (traceId 등)
    Instant occurredAt
) {}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface IdempotentConsumer {
    IdempotencyKey keyType() default IdempotencyKey.EVENT_ID;
    long ttlSeconds() default 86400;   // 24h
}

public enum IdempotencyKey {
    EVENT_ID,       // eventId 기준 (기본 — 이벤트 중복 방지)
    AGGREGATE_ID    // aggregateId 기준 (예: orderId — 집계 단위 중복 방지)
}

@KafkaListener 메서드 호출
        │
        ▼
IdempotencyAspect.around()
        │
        ├── keyType 에 따라 eventId 또는 aggregateId 추출
        │
        ├── Redis SET NX "idempotency:{key}" 1 EX {ttlSeconds}
        │       │
        │       ├── SET 성공 (처음 처리) → proceed() → 실행
        │       │
        │       └── SET 실패 (이미 처리됨) → skip (return, offset commit)
        │
        └── 비즈니스 예외 발생 시 → Redis key 삭제 후 DLT 발행

@KafkaListener(topics = "prd.order.created.v1", groupId = "payment-service")
@IdempotentConsumer(keyType = IdempotencyKey.EVENT_ID, ttlSeconds = 86400)
public void handle(OrderCreatedEvent event) {
    // 중복 eventId 면 AOP 가 이 메서드 자체를 실행하지 않음
    paymentService.processPayment(event);
}

prd.order.created.v1  →  (처리 실패)  →  prd.order.created.v1.DLT

dependencies {
    implementation 'com.example:kafka-common-lib:1.x.x'
}

# application.yml — Soft 설정만 오버라이드
kafka:
  bootstrap-servers: kafka-broker:9092
  producer:
    linger-ms: 50
    batch-size: 32768