# Kafka 연결 테스트
kafka-broker-api-versions.sh --bootstrap-server localhost:9092

spring:
  kafka:
    producer:
      acks: all  # 모든 ISR이 메시지 수신 확인

spring:
  kafka:
    producer:
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5  # idempotence 사용 시 최대 5

spring:
  kafka:
    producer:
      retries: 2147483647          # 무한 재시도 (실제로는 delivery.timeout.ms까지)
      properties:
        retry.backoff.ms: 100      # 재시도 간격
        delivery.timeout.ms: 120000 # 전체 전송 타임아웃 (2분)
        request.timeout.ms: 30000   # 단일 요청 타임아웃

@Component
@RequiredArgsConstructor
@Slf4j
public class ReliableProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final FailedMessageRepository failedMessageRepository;

    public void send(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    handleSendFailure(topic, key, message, ex);
                } else {
                    log.debug("전송 성공: topic={}, partition={}, offset={}",
                        topic,
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                }
            });
    }

    private void handleSendFailure(String topic, String key, String message, Throwable ex) {
        log.error("전송 실패: topic={}, key={}", topic, key, ex);

        // 실패한 메시지를 DB에 저장 (나중에 재처리)
        FailedMessage failedMessage = FailedMessage.builder()
            .topic(topic)
            .messageKey(key)
            .payload(message)
            .errorMessage(ex.getMessage())
            .createdAt(Instant.now())
            .build();

        failedMessageRepository.save(failedMessage);

        // 알림 발송 (선택)
        alertService.sendAlert("Kafka 전송 실패: " + topic);
    }
}

kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders \
  --partitions 6 \
  --replication-factor 3

# 토픽 레벨 설정
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name orders \
  --alter --add-config min.insync.replicas=2

# Broker 전역 설정
min.insync.replicas=2

# server.properties
unclean.leader.election.enable=false  # 기본값: false (Kafka 0.11+)

spring:
  kafka:
    consumer:
      enable-auto-commit: false  # 자동 커밋 비활성화
    listener:
      ack-mode: manual          # 수동 ACK 모드

@KafkaListener(topics = "orders", groupId = "order-processor-group")
public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
    try {
        // 1. 메시지 처리
        processOrder(record.value());

        // 2. 처리 성공 후에만 커밋
        ack.acknowledge();

    } catch (Exception e) {
        log.error("처리 실패: key={}", record.key(), e);
        // 커밋하지 않음 → 재처리됨
    }
}

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
        // 최대 3회 재시도, 1초 간격
        BackOff backOff = new FixedBackOff(1000L, 3);

        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, backOff);

        // 재시도하지 않을 예외 지정
        errorHandler.addNotRetryableExceptions(
            IllegalArgumentException.class,
            JsonParseException.class
        );

        return errorHandler;
    }

    @Bean
    public DeadLetterPublishingRecoverer recoverer(KafkaTemplate<String, String> template) {
        // 재시도 실패 시 DLT(Dead Letter Topic)로 전송
        return new DeadLetterPublishingRecoverer(template,
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
        );
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class IdempotentOrderProcessor {

    private final OrderRepository orderRepository;
    private final ProcessedMessageRepository processedMessageRepository;

    @Transactional
    public void process(String messageId, OrderEvent event) {
        // 1. 이미 처리된 메시지인지 확인 (멱등성 체크)
        if (processedMessageRepository.existsById(messageId)) {
            log.info("이미 처리된 메시지, 스킵: messageId={}", messageId);
            return;
        }

        // 2. 비즈니스 로직 수행
        Order order = Order.from(event);
        orderRepository.save(order);

        // 3. 처리 완료 기록 (같은 트랜잭션)
        processedMessageRepository.save(new ProcessedMessage(messageId, Instant.now()));

        log.info("주문 처리 완료: orderId={}", order.getId());
    }
}

// Producer
public void sendWithTraceId(String orderId, OrderEvent event) {
    String traceId = UUID.randomUUID().toString();

    ProducerRecord<String, String> record = new ProducerRecord<>("orders", orderId, toJson(event));
    record.headers().add("X-Trace-Id", traceId.getBytes(StandardCharsets.UTF_8));
    record.headers().add("X-Timestamp", String.valueOf(System.currentTimeMillis()).getBytes());

    kafkaTemplate.send(record);
    log.info("메시지 전송: traceId={}, orderId={}", traceId, orderId);
}

// Consumer
@KafkaListener(topics = "orders")
public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
    String traceId = new String(record.headers().lastHeader("X-Trace-Id").value());
    log.info("메시지 수신: traceId={}", traceId);

    MDC.put("traceId", traceId);
    try {
        process(record.value());
        ack.acknowledge();
        log.info("메시지 처리 완료: traceId={}", traceId);
    } finally {
        MDC.remove("traceId");
    }
}

# Prometheus Alert Rules
groups:
  - name: kafka-data-loss-alerts
    rules:
      # Producer 전송 실패율
      - alert: HighProducerErrorRate
        expr: rate(kafka_producer_record_error_total[5m]) > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Producer 전송 실패율이 1%를 초과했습니다"

      # ISR 부족
      - alert: InsufficientISR
        expr: kafka_topic_partition_under_replicated > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "복제가 부족한 파티션이 있습니다"

      # Consumer Lag 급증
      - alert: ConsumerLagSpike
        expr: delta(kafka_consumer_lag[5m]) > 10000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Consumer Lag가 급증하고 있습니다"

spring:
  kafka:
    consumer:
      properties:
        max.poll.interval.ms: 600000  # 10분으로 증가
        max.poll.records: 100         # 한 번에 가져오는 레코드 수 감소

spring:
  kafka:
    producer:
      properties:
        max.request.size: 5242880  # 5MB로 증가

# server.properties
message.max.bytes=5242880