결제 시스템으로 알아보는 MSA 3 - 쿼리 문제

예제 소스 코드

이번 블로그에 사용되는 코드는 아래 링크 통해 확인 할 수 있습니다.

https://github.com/syh8088/MSA_project/tree/main/paymentMsa

어떻게 분해를 해야 할까?

분해를 어떤 기준으로 해야 할까요? 앞서 강조한 Business Capability 최대한 가지면서 어떤 방향으로 분해를 해야 할지 고민이 발생 됩니다.

결론은 최대한 Business Capability 유지하면서 분해를 해야 하는데요. 그런데 예를들어 A 서비스 하고 B 서비스 간 통신이 지나치게 발생 된다면 좋은 Business Capability 라고는 할 수 없을 것 입니다.

여기서 대표적으로 2가지 방법으로 분해 기준이 있습니다.

1. 비즈니스 동작
2. 하위 도메인 패턴

이렇게 2가지에 대해 알아보도록 하겠습니다.

비즈니스 동작 기준으로 분리

비즈니스 동작을 기준으로 분리 경우 클라이언트 (사용자) 입장에서 보았을때 비즈니스 요청 별로 각각의 서비스로 분리 하는 것을 비즈니스 동작 기준 분해 라고 합니다.

에시로 결제 시스템을 보았을때 ‘결제 서비스’, ‘정산 서비스’, ‘회원 서비스’ 이렇게 사용자 입장에서 비즈니스 요청 기준 별로 각각의 서비스로 분리 할 수 있겠습니다.

비즈니스 동작 기준으로 분리 할 경우 각각의 분리된 서비스가 미래에 비즈니스 변화가 적을 경우에는 적절 할 수 있는데요. 앞서 예시로 설명 드린 분리된 ‘결제 서비스’ 경우

앞으로 비즈니스 변화가 있다면, 즉 결제 서비스에서 ‘쿠폰 서비스’, ‘배달 서비스’ 등 추가가 되면서 서로 빈번하게 연계되는 상황이 발생한다면 이것은 적절한 서비스 분리가 아닐수도 있겠습니다.

반대로 ‘쿠폰 서비스’, ‘배달 서비스’ 등 추가가 되지 않는다면 적절한 선택 일수도 있겠습니다.

하위 도메인 패턴 기준으로 분리 (from 도메인 주도 설계 - DDD)

하위 도메인 패턴 기준으로 분리 경우 도메인의 복잡성을 이해하고, 도메인의 문제를 해결하는데 집중하여 분리 하는 방식 입니다.

비즈니스 요구 사항의 변화가 많을 경우에는 이에 대응 하여 Business Capability 수준을 적절하게 대응 하기 위해서는 하위 도메인 패턴 기준으로 분리 가 적절 할 수도 있겠습니다.

앞써 동일한 예시로 기존에 비즈니스 동작 기준 으로 분리 한 ‘결제 서비스’, ‘정산 서비스’, ‘회원 서비스’ 로 설명하자면 이 중에 ‘결제 서비스’ 는
적절한 Business Capability 요구사항을 충족하기 위해서 ‘쿠폰 서비스’, ‘배달 서비스’ 으로 파생되어서 하위 도메인 기준으로 분리 할 수 있습니다.

결국은 비즈니스를 식별 하는 것은 가장 우선순위는 분리된 각각의 서비스가 적절하게 Business Capability 요구 사항에 적절한지를 체크 하는것이 중요하다고 볼 수 있겠습니다.

여기서 나오는 도메인 주도 설계 (DDD, Domain-Driven Design) 설명

도메인 주도 설계 (DDD, Domain-Driven Design) 는 하나의 소프트웨어 개발 방법론으로, 복잡한 도메인을 이해하고 해결하는데 목적을 가지고 이에 집중하는 방법론 입니다.

도메인 주도 설계 나타나기 전에는 도메인이 복잡해지면서 발생하는 문제들이 많았었는데 이에따라 유지보수가 어렵다는 문제를 해결하기 위해 탄생된 개발 방법론 이라고 보면 될 것 같습티다.

MSA 진행하는데 있어서 분리 하는 과정에 도메인 주도 설계 하고 무슨 관계가 있을까? 생각 할 수도 있는데 도메인 주도 설계 핵심 원칙인 Bounded Context 와 Sub Domain 개념이 MSA 목적과 같기 때문 입니다.

즉 도메인 주도 설계 (DDD, Domain-Driven Design) 의 핵심 개념들은 MSA 의 책심 개념이었던 개별 팀이 Product 로써 서비스를 소유한다는 사고 방식과 같다는 것 입니다.

도메인 주도 설계 (DDD, Domain-Driven Design) 에서 중요한 키워는 Bounded Context 와 Sub Domain 존재한다고 했습니다.

여기서 말하는 Bounded Context 는 정의한 여러 도메인들 기준으로 각 도메인 모델들의 관계 표현 시 각 모델들은 상호 배타적인 둔제를 해결해야만 한다는 도메인 주도 설계 DDD 의 핵심원칙 입니다.

이미지를 보시면 ‘주문 도메인’ 경우 ‘주문’, ‘결제’, ‘쿠폰’ 로 분리 된다고 가정시 각각의 모델 영역에서 명확한 Bounded 를 가지고 분리 될 수 있겠습니다.
각각의 ‘주문’, ‘결제’, ‘쿠폰’ 모델은 하나의 기능을 집중 하므로써 서로 배타적인 문제를 가지는 것이 Bounded Context 의 핵심 원칙 입니다.

다음으로 설명하는 Sub Domain 경우는 이렇게 정의한 도메인 도델들이 Bounded Context 원칙에 따라서 Business Capability 를 높이도록 설계 및 분리 하는 원칙 이라고 보면 됩니다.
다르게 말씀드리자면 비즈니스 동작으로 부터 식별된 동작을 Business Capability 요구에 맞게 분리된것을 말 합니다. 이렇게 분리된 Bounded Context 원칙에 따라서 분리 해야 합니다.

분해로 인해 발생되는 문제 - 트랜잭션

기존 모놀리스 방식에서 트랜잭션 구현은 쉽고 간단합니다. DB 데이터 베이스 직관적으로 연계되고 있기 때문에 관리 포인트가 비교적 쉽습니다.
하지만 MSA 에서 트랜잭션 구현은 굉장히 어렵고 관리 포인트가 많습니다.
각 서비스 마다 각각의 데이터베이스로 분리되어 있는 MSA 환경은 트랜잭션을 구현하기 위해서는 분산 트랜잭션을 구현 할 수 있어야 합니다. 이를 위해
2PC(2 Phase Commit), 보상 트랜잭션(Compensating Transaction), SAGA 같은 여러가지 방법으로 해결 합니다.

2PC(2 Phase Commit) 란?

초기에는 다수의 데이터베이스 노드 환경에서 커밋을 구현하기 위해 나타났습니다. 점점 발전되어 분산 시스템 환경에서 트랜잭션을 구현하기 위해 대표적인 방법으로 발전 해왔습니다.

2PC 를 구현 하기 위해서는 분산 시스템 환경에 Cordinator 라는 추가적인 인프라가 필요하고, 트랜잭션을 위한 서비스들을 Participant (참여자) 이라고 말합니다.

여기서 말하는 Cordinator 역활은 MSA 같은 분산 서버에서 분산 트랜잭션을 구현하기 위해 중앙에 관리 해주는 역활을 하고 있습니다.

// 이미지

Cordinator Participant-1, Participant-2, Participant-3 이렇게 있다고 하면 가장 먼저 트랜잭션이 시작되면 Cordinator 가 Global Unique ID (Transaction ID) 를 생성 하게 됩니다.

그런 다음 Cordinator 는 Participant-1 ~ Participant-3 까지 prepare request 요청을 합니다. 이때 트랜잭션이 시작과 동시에 동시성을 보장 하기 위해 Lock 시작 하게 됩니다.
각각의 Participant 들이 트랜잭션을 처리 후 DB 업데이트 한 이후에 준비 가 되었다고 Cordinator 에게 Vote 를 하게 됩니다.

여기까지는 전체적인 Prepare 단계이고 그 다음 단계는 Cordinator 는 해당 Vote 계산해서 모두 정상적으로 commit 해도 된다는 Commit Intialized 신호를 전달 하게 됩니다. 그런 후 Participant 들은 commit 처리 하게 완료가 됩니다.

하지만 특정 Participant 에서 DB 업데이트 하는 과정에서 문제가 발생 시 Cordinator 에게 준비 하는 과정에서 문제가 발생 했다고 전달 하게 되는데요.
이를 전달 받은 Cordinator 은 Participant 들에게 Abort 신호를 전달해서 처리 하게 됩니다.

하지만 2PC 가 보편적으로 활용되지 않는 여러 이유가 있는데 Cordinator 가 Participant 들에게 Request to prepare 신호를 전달 시
Participant 들은 Lock 을 걸고 Cordinator 에게 준비 되었다고 전달 할때 Cordinator 서버가 문제가 발생 되어 통신을 할 수 없는 상황이 온다면
Lock 을 걸어놓고 멈춰버리는 문제가 발생 됩니다.

보상 트랜잭션(Compensating Transaction) 란?

// 이미지 준비

보상 트랜잭션는 MSA 같은 분산 시스템 환경에서 트랜잭션을 유지하기 위한 방법 중 하나 입니다.
DBMS 기준으로 commit 된 데이터를 commit 되기 이전의 상태로 변경하는 작업 을 보상 트랜잭션 이라고 말합니다.

그럼 MSA 환경에서는 특정 API 호출로 인해 commit 된 데이터를 어느 문제가 발생되어서 commit 되기 이전의 상태로 되돌아 가기 위해
되돌리기 위한 API 를 호출 하므로써 이전의 데이터 변경 요청을 하게 됩니다.

이러한 방식으로 개발시 commit 되기전 상태로 만들기 위한 API 개발도 필요하기 때문에 작업량을 더 늘어난다는 것 입니다.
이때 고려해야 되는 부분은 보상 트랜잭션 요청시 지연 혹은 실패 문제가 생기는 경우 비즈니스 상황에 맞는 후속 처리가 필요하게 됩니다.

SAGA 패턴

앞서 말씀드린 2PC, 보상 트랜잭션 에서 분산 시스템 환경에서 분산 트랜잭션을 구현하기 위한 방법으로 소개 해드렸습니다.
하지만 각각의 방법에서 단점이 존재 합니다. 모든 기술에 트레이드 오프가 존재하지만 SAGA 패턴 에 대해서도 설명 하고자 합니다.

SAGA 패턴은 이벤트 방식의 트랜잭션에 포함된 여러 작업들을 결과를 제시하고 이벤트를 이용해 비동기로 처리하는 방식 입니다. 앞써 이야기한 2PC, 보상 트랜잭션 장점을 가져와서 만든 패턴인데요.

예를들어 ‘작업1~ '작업3' 이 존재한다면작업 1` 을 실행 후 성공된다면 성공 이벤트 발행 하게 됩니다. ‘작업2’ 는 consume 해서 해당 작업을 실행 해서
같은 방법으로 ‘작업3’ 까지 진행하게 됩니다. 이미지를 보면서 자세하게 설명 하도록 하겠습니다.

A 서비스 부터 C 서비스까지 존재시 순서는 A -> B -> C 까지 각각의 오퍼레이션을 실행 합니다. 처음에 A 서비스가 실행 동시에 트랜잭션 시작을 하게 되고 작업이 끝나면 B 서비스에 작업 시작 이벤트를 의미 하는
이벤트 메시지를 발행 합니다. B 서비스는 consume 해서 트랜잭션을 시작 후 동일한 방법으로 작업이 끝나면 C 서비스에 전달 하기 위한 이벤트 메시지를 발행 합니다.

만약 A 서비스 부터 시작하다가 C 서비스 에서 Exception 이 발생 된다면 C 서비스는 B 서비스에게 rollback 하기 위한 메시지 이벤트를 발행 하게 되고

동일한 방법으로 A 서비스까지 이벤트 메시지 통해 전달 하게 됩니다. 즉 이벤트 기반으로 성공 및 실패를 주거니 받거니 하고 있습니다.

SAGA 패턴 을 구현하기 위해 대표적으로 방법 2가지가 있는데요.

코레오그레피 (Choreography) 패턴 오케수트레이션 (Orchestration) 패턴 이 있습니다.

코레오그레피 (Choreography) 패턴 는 독립적인 조율자(Ochestrator) 를 두지 않고, SAGA 를 구현하는 방식 입니다. 매우 간단 하지만 트랜잭션 상황에 모니터링 하기가 어렵다는 단점이 있습니다.

이와 반대로 오케수트레이션 (Orchestration) 패턴 는 독립적인 조율자(Ochestrator) 를 두어 하나의 Saga (트랜잭션) 에 대한 매니징을 담당하는 방법이에요.
구현이 어렵지만, 모니터링 하는데 있어서 수월하다는 장점이 있습니다.

SAGA 패턴 - 오케수트레이션 (Orchestration) 패턴

SAGA 패턴에서 오케수트레이션 (Orchestration) 패턴 경우 메시지를 발행하고 그에 대한 결과를 수신 받는 주체는 조율자(Ochestrator) 입니다.

1. 조율자(Ochestrator) 통해 ‘B 서비스’ 에게 작업을 수행 하기 위해 메시지를 발행을 합니다.
2. 수신 받은 B 서비스는 해당 작업을 수행 합니다.
3. 수행 후 결과 응답을 메시지 발행 합니다.
4. 조율자(Ochestrator) 는 결과 메시지 수신 받습니다.
5. C 서비스에 작업을 수행하기 위해 메시지를 발행 합니다.
6. C 서비스에 작업 수행 메시지 받고 작업을 수행 합니다.
7. 작업을 수행 한 후 작업 결과 메시지를 발행 합니다.
8. 조율자(Ochestrator) 는 C 서비스 작업 결과 메시지를 수신 받습니다.
9. 마지막 A 서비스 작업을 수행하기 위해 작업 메시지 발행 합니다.
10. A 서비스는 작업 메시지를 받고 작업을 수행 합니다.

1. C 서비스까지 작업이 실행되다가 작업 실행 도중 실패가 발생되어서 실패 이벤트 발행 합니다.
2. 조율자(Ochestrator) 에게 C 서비스 실패 메시지 수신 받게 됩니다.
3. 조율자(Ochestrator) 는 B 서비스에게 보상 트랜잭션을 실행 하기 위해 메시지 발행 합니다.
4. B 서비스는 실패로 인한 메시지를 받아서 보상 트랜잭션을 실행 합니다.
5. 조율자(Ochestrator) 는 A 서비스에게도 실패 인한 메시지 발행 합니다.
6. A 서비스는 실패 메시지를 수신 받고 이에 따른 응답을 하게 됩니다.

결제 시스템에서 분산 트랙잭션이 필요한 구간 알아보기

분산 시스템에서 분산 트랜잭션을 구현 하는 것은 가능한 해당 비지니스가 분산 트랜잭션을 구현을 꼭 해야 되는 상황인지 판단 하는 것이 좋습니다.
만약 굳이 분산 트랜잭션을 구현하지 않더라도 기획단계에서 끝낼수 있는 부분이라면 분산 트랜잭션을 구현하지 않고 진행하는 것도 좋은 방법 중 하나 일 것 입니다.

클라이언트가 결제 시도를 하게 되면 주문 서비스에서 PG사 Toss 서버 통해 결제 승인이 발생 됩니다. 결제 결과에 따라 결제 상태 업데이트를 진행하게 되고
그 다음 정산을 진행 하기 위해 이벤트 메시지를 발행과 동시에 고객에게 결제 결과를 응답 하게 됩니다.

만약에 정산 서비스에서 문제가 발생 된다면 주문 서비스에서 결제 결과에 따른 DB 업데이트를 모두 보상 트랜잭션 통해 롤백 해야 할까요?
이미 결제 승인이 발생된 부분이기 때문에 롤백은 고려 해야 될 부분 입니다. 정산 경우는 지금 당장 실행을 안해도 괜찮은 비지니스 이라면 굳이 롤백 보다는 DLQ(Dead Letter Queue) 에
보관을 하고 나중에 조치를 하면 될 것 같습니다. 그리고 또한 이러한 경우는 개발자에게 알림 메시지 통해 인지 해주는 것이 중요 하다고 생각 됩니다.

즉 서비스상 굳이 분산 트랜잭션을 구현 할 필요없이 다른 방향으로 해결이 가능하면 가장 좋은 방향이고, 어쩔수 없이 구현 해야 한다면 분산 트랜잭션을 구현 해야 할 것입니다.

여기서는 Seller 가 초기 등록되면 이에 해당되는 Seller 의 Wallet 을 등록 하도록 합니다. 만약 실패가 된다면 등록된 Seller 를 상태 변경 업데이트를 진행 하고자 합니다.

Axon 이용해서 Saga 패턴으로 분산 트랜잭션 구현 해보자

Axon 을 이용해서 Saga 패턴을 사용하면 쉽게 오케수트레이션 (Orchestration) 패턴 을 구현 하고자 합니다.

여기서 Axon Framework 는 EventSourcing, CQRS, DDD 패턴 중심으로 서비스를 쉽게 구현 할 수 있도록 도움을 줍니다. 더불어 Axon Server 는
Axon Framework 에 여러 기능들을 작동 할 수 있도록 해주는 서버 입니다. 즉 ‘Axon Server’ 는 메세지 라우팅, 이벤트 저장소 기능을 이용해서 Axon Framework 는 Axon Server 에 지원하는
기능을 이용해 EventSourcing, CQRS 등 여러 기능을 제공 합니다.

이미지에 보시다 싶이 Axon Framework 하고 Axon Server 조합을 이용해서 기능을 구현하는데요. Axon Server 대신 다른 Event Broker 인 Kafka 등 이용 해도 괜찮습니다.

여기서는 Axon Framework, Axon Server Saga 오케수트레이션 (Orchestration) 패턴 을 구현 해보도록 하겠습니다.

Seller 서비스하고 Wallet 서비스는 각각 Axon Framework 을 가지고 있습니다. 이 두개의 서비스는 Axon Server 와 연계 해서 Axon Server 에서 지원하는 Queue-Louting, Event Store 기능 이용해
이벤트 소싱을 위한 EDA 를 구축하고 이를 바탕으로 Saga 오케수트레이션 (Orchestration) 패턴 을 구현 하겠습니다.

클라이언트가 Seller 를 등록 하게 된다면 Seller 서버는 Axon Server 에 있는 Event queue 로 이벤트 발행 합니다. 여기서 Axon Server 내부에서 자체적으로 Consume 하게 되고 그 다음
Event 저장을 하기위해 Event Store 에 이벤트를 저장 하게 됩니다. 그 다음 Event Sourcing 진행하게 되면서 Seller 서비스는 디비 서버와 연계해 Seller 를 등록 하게 됩니다.

이렇게 아키텍처를 보았을때 의문이 가는 부분은 초기 클라이언트가 Seller 서버로 요청이 들어올때 DB 서버에 바로 Seller 를 등록하지 않고 Axon Server 내에 Event Queue 에 등록 하게 될까요?
Event Queue 로 등록 해서 관리 하는 것이 신뢰성 및 가용성 측면으로 보았을때 더 높기 때문입니다.

일단 Axon Server 내에 Event Queue 에 해당 Command 를 저장 하기만 한다면 이후에 전체적인 서버 문제가 발생 해서 Seller 등록 진행이 중단 되었을때 이미 Event Queue 에 등록되어 있으니
나중에 복구 후 다시 시도 하면 문제가 없을 것 입니다.

Seller 등록 설계 - 성공

클라이언트 로부터 Seller 등록 요청이 들어오면 Axon Server 에 이벤트 발행 하게 되고 Axon Server 는 Seller 서비스에 전달해서 Seller 를 DB 에 등록 하게 됩니다.

그다음 등록된 Seller 에 해당 되는 Wallet 등록 하기 위해 Axon Server 에 요청해서 Wallet 서비스는 디비에 Wallet 을 등록 하게 됩니다.

Seller 등록 설계 - 실패

만약 Wallet DB 에 등록하는 과정에서 에러가 발생된다면 Seller Server 에 보상 트랜잭션 을 보장 하기 위해 Axon Server 에 요청해서 Seller 서비스는
Wallet 등록 실패로 인한 디비 업데이트를 진행 하게 됩니다.

구현 코드

@Slf4j
@UseCase
@RequiredArgsConstructor
public class SellerService implements InsertSellerUseCase {

    private final CommandGateway commandGateway;

    @Override
    public void insertSeller(RequestPersistenceInsertSellerCommand insertSellerCommand) {

        RequestInsertSellerCommand requestInsertSellerCommand = RequestInsertSellerCommand.of(insertSellerCommand.getSellerId());
        commandGateway.send(requestInsertSellerCommand)
        .whenComplete(
                (result, throwable) -> {
                    if (throwable != null) {
                        throwable.printStackTrace();
                        log.error("InsertSellerCommand Command failed = {}", result.toString());
                        throw new RuntimeException(throwable);
                    }
                    else {
                        log.info("InsertSellerCommand SAGA success = {}", result.toString());
                    }
                }
        );
    }
}

AxonServer 의 CommandBus 에 Command (RequestInsertSellerCommand: Seller 등록) 를 보냅니다.

@Aggregate
@Data
public class SellerAggregate {

    @AggregateIdentifier
    private String id;

    @CommandHandler
    @CreationPolicy(AggregateCreationPolicy.ALWAYS)
    public void handler(RequestInsertSellerCommand command) {

        this.id = command.getId();
        String requestInsertSellerId = UUID.randomUUID().toString();

        // Saga Start
        apply(new RequestInsertSellerEvent(requestInsertSellerId, command.getSellerId()));
    }

}

Seller 의 Aggregate 는 요청 받은 Command 를 받아 Seller 등록 Event 를 발행 합니다.

@Slf4j
@Saga
@NoArgsConstructor
public class InsertSellerAndWalletSaga {

    @StartSaga
    @SagaEventHandler(associationProperty = "requestInsertSellerId")
    public void handle(RequestInsertSellerEvent event) {

        String sagaInsertSellerId = UUID.randomUUID().toString();
        SagaLifecycle.associateWith("sagaInsertSellerId", sagaInsertSellerId);

        commandGateway.send(
                new InsertSellerCommand(
                        UUID.randomUUID().toString(),
                        event.getRequestInsertSellerId(),
                        event.getSellerId(),
                        sagaInsertSellerId
                )
        ).whenComplete(
                (result, throwable) -> {
                    if (throwable != null) {
                        throwable.printStackTrace();
                        log.error("RequestInsertSellerEvent Command failed, = {}", sagaInsertSellerId, throwable);
                    } else {
                        log.info("RequestInsertSellerEvent SAGA success = {}", result.toString());
                    }
                }
        );
    }
}

등록 Seller 담당하는 InsertSellerAndWalletSaga 는 해당 Event 를 구독하여 트리거 됩니다.

메소드 상단에 @StartSaga 확인 할 수 있습니다. 이는 Saga 가 시작 되었다는 것을 알립니다.

@SagaEventHandler(associationProperty = "requestInsertSellerId")

해당 로직은 구독할 Saga 를 정하고 식별자를 지정하는 용도 입니다.

@Slf4j
@Aggregate
@Data
public class InsertSellerAggregate {

    @AggregateIdentifier
    private String id;

    @CommandHandler
    @CreationPolicy(AggregateCreationPolicy.ALWAYS)
    public void requestInsertWalletAggregate(
            InsertSellerCommand command,
            InsertSellerPort insertSellerPort
    ) {
        id = command.getId();

        // seller insert
        long savedSellerNo = insertSellerPort.insertSeller(
                RequestPersistenceInsertSellerCommand.of(command.getSellerId())
        );

        String requestInsertWalletId = UUID.randomUUID().toString();

        // Saga Start
        apply(new SagaInsertWalletEvent(
                requestInsertWalletId,
                command.getSagaInsertSellerId(),
                command.getRequestInsertSellerId(),
                savedSellerNo,
                command.getSellerId()
        ));
    }

}

등록 담당 하는 Seller 의 Aggregate 는 요청 받은 Command 를 받아 DB 에 Seller 를 insert 하게 됩니다.

그런 후 그다음 로직을 수행 해야 할 해당 Seller 의 Wallet 등록을 하기 위해 Event 를 발행 합니다.

@SagaEventHandler(associationProperty = "sagaInsertSellerId")
public void handle(SagaInsertWalletEvent event) {

    String sagaInsertWalletId = UUID.randomUUID().toString();
    SagaLifecycle.associateWith("sagaInsertWalletId", sagaInsertWalletId);

    commandGateway.send(new InsertWalletCommand(
            UUID.randomUUID().toString(),
            sagaInsertWalletId,
            event.getSagaInsertSellerId(),
            event.getSellerNo(),
            event.getSellerId()
        )
    ).whenComplete(
            (result, throwable) -> {
                if (throwable != null) {
                    throwable.printStackTrace();
                    log.error("requestInsertWalletAggregate Command failed");
                } else {
                    log.info("requestInsertWalletAggregate SAGA success = {}", result.toString());
                }
            }
    );
}

Saga 는 CommandGateway 를 이용해서 Wallet 을 등록하는 Command 를 보냅니다.

@Slf4j
@Aggregate()
@Data
public class InsertSellerRequestInsertWalletAggregate {

    ... 생략 ...

    @CommandHandler
    @CreationPolicy(AggregateCreationPolicy.ALWAYS)
    public void requestInsertWalletAggregate(
            InsertWalletCommand insertWalletCommand,
            InsertWalletPort insertWalletPort
    ) {

        id = insertWalletCommand.getAggregateIdentifier();

        // wallet insert
        boolean isSuccess = insertWalletPort.insertWallet(com.wallet_service.domain.InsertWalletCommand.of(insertWalletCommand.getSellerNo()));
        
        String requestResultInsertWalletId = UUID.randomUUID().toString();

        apply(new RequestInsertWalletFinishedEvent(
                requestResultInsertWalletId,
                insertWalletCommand.getSagaInsertWalletId(),
                insertWalletCommand.getSagaInsertSellerId(),
                isSuccess,
                insertWalletCommand.getSellerNo(),
                insertWalletCommand.getSellerId()
        ));
    }

    ... 생략 ...
}

등록 담당 하는 Wallet 의 Aggregate 는 요청 받은 Command 를 받아 DB 에 Wallet 을 insert 하게 됩니다.

그런 후 만약 등록 성공 여부 데이터 값 함께 Event 를 발행 합니다.

@SagaEventHandler(associationProperty = "sagaInsertWalletId")
public void handle(RequestInsertWalletFinishedEvent event) {

    boolean isSuccess = event.isSuccess();
    if (isSuccess) {
        SagaLifecycle.end();
    }
    else {
        // 실패 시, 롤백 이벤트
        String sagaRollbackSellerId = UUID.randomUUID().toString();
        SagaLifecycle.associateWith("sagaRollbackSellerId", sagaRollbackSellerId);
        commandGateway.send(
                new RollbackSellerRequestCommand(
                    sagaRollbackSellerId,
                    event.getSellerNo()
                )
        ).whenComplete(
                (result, throwable) -> {
                    if (throwable != null) {
                        throwable.printStackTrace();
                        log.error("RequestWalletFinishedEvent Command failed");
                    }
                    else {
                        log.info("RequestWalletFinishedEvent SAGA success = {}", result.toString());
                        SagaLifecycle.end();
                    }
                }
        );
    }
}

최종 Wallet 등록까지 성공적이면 Saga 를 종료 하지만 만약 실패 할 경우 롤백 (여기서는 정확히 해당 Seller 상태값 변경)을 실행 해야 합니다.

그래서 Saga 는 CommandGateway 이용해서 롤백 로직을 담당하는 Command 를 보냅니다.

@CommandHandler
@CreationPolicy(AggregateCreationPolicy.ALWAYS)
public void requestInsertWalletAggregate(
        RollbackSellerRequestCommand command,
        DeleteSellerPort deleteSellerPort
) {

    id = command.getSagaRollbackSellerId();

    // seller delete
    deleteSellerPort.deleteSeller(command.getSellerNo());

    String requestDeleteSellerId = UUID.randomUUID().toString();

    // Saga Start
    apply(new RequestDeleteSellerEvent(requestDeleteSellerId, command.getSagaRollbackSellerId(), command.getSellerNo()));
}

Seller 담당하는 Aggregate 는 해당 Seller 롤백 처리 (여기서는 간단하게 Seller 상태값 변경)를 하고 롤백 완료 Event 를 발행 하도록 합니다.

@EndSaga
@SagaEventHandler(associationProperty = "sagaRollbackSellerId")
public void handle(RequestDeleteSellerEvent event) {
    log.info("EndSaga >> requestDeleteSellerId = {}", event.getRequestDeleteSellerId());
}

Saga 가 Event 를 받으면 @EndSaga 따라 Saga 를 종료하게 됩니다.

Reference

https://medium.com/@blogs4devs/saga-with-axon-bb7b9cfe0b64
https://github.com/Blogs4Devs/SAGA-Axon
https://jaehoney.tistory.com/403

---

Copyright 201- syh8088. 무단 전재 및 재배포 금지. 출처 표기 시 인용 가능.