TIL - DB/Redis/Kafka 분산트랜잭션 대응

코드 리뷰

이번에 최종프로젝트를 진행하는 다른 조의 코드를 리뷰해봤다.
분산 트랜잭션 문제를 해결하기 위한 고민들을 엿보며 어떤 기술들을 왜 채택했는 지 알아보았다.

Problem

기존에는 Match 엔티티의 상태를 바꾸면 됐다.

@Transactional
    public MatchResult changeStatus(ChangeMatchStatusCommand command) {
        Match match = matchRepository.findByIdAndDeletedAtIsNull(command.matchId())
                .orElseThrow(() -> new MatchNotFoundException(command.matchId()));
        match.changeStatus(command.targetStatus());
        return MatchResult.from(match);
}

그런데 상태를 바꾸면서 redis에 저장도하고
kafka로 이벤트도 발행해야 한다면?

@Transactional은 JPA에서 관리하기 때문에 실패 시, db 저장만 롤백해준다.
그래서 db저장은 롤백되지만 redis에는 저장되고 kafka로 이벤트가 발행되는 문제가 생길 수 있다.

Analize

Facade패턴을 적용해서 MatchApplicationService에서는 matchWriteService의 메서드를 호출만 하고 내부 로직은
MatchWriteService에서 처리하도록 했다.

@Transactional
public MatchResult changeStatus(ChangeMatchStatusCommand command) {
    return matchWriteService.changeStatus(command);
}

 

그랬더니 MatchWriteService가 도메인 로직과 외부 인프라 통신 로직을 모두 가져가면서
단일 책임 원칙(SRP)을 위반하고 코드가 비대해졌다.

또한 DB와 외부 시스템(Redis, Kafka)은 분산 시스템 환경이므로 하나의 트랜잭션으로 묶일 수 없다.
따라서 DB 커밋 성공 여부에 따라 후속 작업이 실행되도록 제어할 필요성이 있었다.

Action

먼저 관심사를 분리하여 서비스 구현체에서 Match 상태 변경이라는 핵심 도메인 로직을 먼저 수행하고,
부가적인 로직을 호출하도록 분리하여 응집도를 높였다.

@Transactional
public MatchResult changeStatus(ChangeMatchStatusCommand command) {
	Match match = getActive(command.matchId());
        match.changeStatus(command.targetStatus());
    return matchWriteService.changeStatus(command);
}

 

그리고 트랜잭션 동기화(TransactionSynchronizationManager.registerSynchronization)를 통해 DB 커밋이 최종 완료된 경우에만 Redis 초기화 로직이 실행되도록 afterCommit 핸들러를 구현했다.

마지막으로 Kafka 메시지 유실을 방지하기 위해 Transactional Outbox 패턴을 도입하여 DB와 이벤트 발행의 원자성을 확보했다.

public MatchResult changeStatus(ChangeMatchStatusCommand command) {
        if (command.targetStatus() == MatchStatus.APPROVED) {
            matchEventPublisher.publishMatchApproved(
                    new MatchApprovedEvent(match.getId(), match.getTicketOpenAt())
            );
            UUID matchId = match.getId();
            Map<UUID, Long> seatCounts = match.getZonePolicies().stream()
                    .filter(p -> p.getDeletedAt() == null)
                    .collect(Collectors.toMap(
                            MatchZonePolicy::getSeatGradeId,
                            MatchZonePolicy::getTotalSeatCount,
                            (existing, duplicate) -> {
                                log.warn("[MatchWriteService] matchId={} — seatGradeId 중복 ZonePolicy 감지. "
                                        + "도메인 불변식 위반 가능성. 첫 번째 값({}) 사용.", matchId, existing);
                                return existing;
                            }
                    ));
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    try {
                        seatAvailabilityRepository.initialize(matchId, seatCounts);
                        log.info("[MatchWriteService] matchId={} Redis 잔여 좌석 초기화 완료. zones={}",
                                matchId, seatCounts.size());
                    } catch (Exception e) {
                        log.error("[MatchWriteService] matchId={} Redis 잔여 좌석 초기화 실패. "
                                + "APPROVED 커밋은 완료됐으나 Redis 가 비어 있음. "
                                + "수동 재초기화 또는 재승인 처리 필요.", matchId, e);
                    }
                }
            });
        } else if (command.targetStatus() == MatchStatus.CANCELED) {
            matchEventPublisher.publishMatchCanceled(
                    new MatchCanceledEvent(match.getId(), OffsetDateTime.now())
            );
        }

        return MatchResult.from(match);
    }

 

Result

DB 트랜잭션 성공 여부에 따른 작업의 최종적 일관성(Eventual Consistency)을 확보했고,
실패 시에도 에러 로그와 멱등성 설계를 통해 수동 재처리가 가능하도록 구현하여 시스템의 안정성과 신뢰성을 높였다.