Kafka 파티션 리밸런싱중 발생하는 중복 메시지 처리, 멱등성 보장

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.dao.DataIntegrityViolationException;

@Service
public class PaymentConsumerService {

    private final PaymentRepository paymentRepository;
    private final IdempotencyKeyRepository idempotencyKeyRepository;

    public PaymentConsumerService(PaymentRepository paymentRepository, IdempotencyKeyRepository idempotencyKeyRepository) {
        this.paymentRepository = paymentRepository;
        this.idempotencyKeyRepository = idempotencyKeyRepository;
    }

    @Transactional
    public void processPaymentMessage(PaymentMessage message) {
        String idempotencyKey = message.getOrderId();

        try {
            // 1. 멱등성 키 저장 시도 (DB의 Unique Key 제약 조건 활용)
            // 이미 존재하는 키라면 DataIntegrityViolationException 발생
            IdempotencyRecord record = new IdempotencyRecord(idempotencyKey, "PROCESSING");
            idempotencyKeyRepository.saveAndFlush(record);

            // 2. 실제 비즈니스 로직 수행 (결제 승인 등)
            doPayment(message);

            // 3. 상태 업데이트
            record.setStatus("COMPLETED");

        } catch (DataIntegrityViolationException e) {
            // 중복 메시지 수신: 리밸런싱이나 재시도로 인한 중복이므로 무시하고 정상 종료(Ack) 처리
            // 주의: 여기서 예외를 던지면 무한 재시도(Poison Pill)에 빠질 수 있습니다.
            System.out.println("Duplicate message ignored for key: " + idempotencyKey);
        } catch (Exception e) {
            // 기타 비즈니스 예외 발생 시에는 트랜잭션을 롤백하고 적절한 에러 처리 (DLQ 전송 등) 수행
            throw e;
        }
    }
}

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Duration;

@Service
public class PaymentConsumerRedisService {

    private final StringRedisTemplate redisTemplate;
    private final PaymentService paymentService;

    public PaymentConsumerRedisService(StringRedisTemplate redisTemplate, PaymentService paymentService) {
        this.redisTemplate = redisTemplate;
        this.paymentService = paymentService;
    }

    public void processPaymentMessage(PaymentMessage message) {
        String idempotencyKey = "idemp:payment:" + message.getOrderId();
        
        // 1. Redis SETNX를 활용한 원자적 락 획득 (TTL 24시간 설정)
        // 키가 없으면 저장하고 true 반환, 이미 있으면 false 반환
        Boolean isFirstRequest = redisTemplate.opsForValue()
                .setIfAbsent(idempotencyKey, "PROCESSING", Duration.ofHours(24));

        // 2. 중복 검증 로직
        if (Boolean.FALSE.equals(isFirstRequest)) {
            // 이미 다른 컨슈머(또는 이전 리밸런싱 전 스레드)가 처리 중이거나 완료한 메시지
            System.out.println("Duplicate message filtered by Redis: " + idempotencyKey);
            return; // 조용히 Ack 처리하고 종료 (DB 접근 원천 차단)
        }

        try {
            // 3. 실제 비즈니스 로직 및 DB 반영 (별도의 트랜잭션으로 분리)
            paymentService.executePaymentLogic(message);
            
            // 4. 처리 완료 상태로 업데이트 (선택 사항)
            redisTemplate.opsForValue().set(idempotencyKey, "COMPLETED", Duration.ofHours(24));
            
        } catch (Exception e) {
            // 예기치 않은 서버 에러 시, 재시도를 위해 Redis 락 해제
            redisTemplate.delete(idempotencyKey);
            throw e;
        }
    }
}

kafka-consumer-groups.sh \
  --bootstrap-server kafka-broker01:9092 \
  --group payment-consumer-group \
  --describe

GROUP                  TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                     HOST            CLIENT-ID
payment-consumer-group payment-events  0          10050           10050           0               consumer-payment-1-xyz  /10.0.1.10      consumer-payment-1
payment-consumer-group payment-events  1          20010           25000           4990            consumer-payment-2-abc  /10.0.1.11      consumer-payment-2
payment-consumer-group payment-events  2          15000           15000           0               -                                               -               -