feat: impl orchestration saga pattern

[O0oO0Oo]

PR 설명

오케스트레이셔 사가 패턴의 구현.

변경 사항

• saga-orchestrator 모듈 추가
• saga 구현체들
  • SagaStep : 각 사가 단계에서 진행 될 실행 동작 정보 클래스, 이전 단계들의 응답을 바탕으로 등록한 Consumer 를 통해, MessageProducer 로 다른 서비스로의 요청이 나간다.
  • SagaPromise : 각 단계에서 Netty 의 Promise 를 사용하여 응답이 오면 setSuccess 를 설정하고, addlistener 를 통해 다음 SagaPromise 의 SagaStep 을 실행
  • SagaPromiseAggregator : 이전 단계가 병렬적으로 진행되고, 이전의 복수 단계들이 성공했을때 스텝이 진행되도록 설정
  • SagaPromiseFactory : SagaPromise 를 생성하는 팩토리 Step 타입에 따라 SagaPromiseCreationStrategy 를 선택한다
  • SagaPromiseCreationStrategy : SagaPromise 생성 전략 패턴, ServiceLoader 를 사용하여 SagaStep 의 종류에 따른 SagaPromise 가 생성된다,
  • SagaDefinition : 사가의 각 단계 순서와 동작을 정의, initializeSaga() 호출 시 SagaPromise 체인 리스트를 생성하여 제공
  • SagaCoordinator : SagaDefinition, SagaStateManager 를 등록하고 MessageConsumer 를 등록하면 클라이언트의 요청으로 사가의 시작과 구독한 응답 토픽을 바탕으로 SagaStateManager 에게 SagaState 를 업데이트하거나 시작하는 요청을 한다.
  • SagaStateManager : SagaState 를 관리하며, 응답으로 온 Message 객체에서 correlationId, stepId 를 바탕으로 SagaState 의 특정 단계 성공 그리고 실패시 보상 요청을 하도록 한다.
  • SagaState : SagaPromise 체인리스트를 담고있다. 각 응답을 저장 또는 failure 로 설정하여 보상 트랜잭션 요청을 실행하도록 할 수 있다.

배경

오케스트레이션 사가를 구현하면서 다음과 같은 요구사항이 있었습니다.

1. 모든 작업은 비동기적으로 진행
2. 각 작업은 이벤트로 온 응답값으로 완료된다.
3. 여기서 각 작업은 이 응답값을 바탕으로 이벤트를 발행하여 다른 서비스로 요청하는 것이다.
4. 각 작업은 이전의 작업들이 완료되어야 진행될 수 있다. (이전의 작업은 하나 또는 그 이상)
5. 보상 트랜잭션 이벤트 발행을 위한, 이전 결과값을 재활용 해야한다.

이러한 요구사항을 위해 다음과 같은 정보를 저장해야 했습니다.

1. 이벤트로 온 응답값
2. 각 작업의 성공 상태
3. 현재 작업이 실행되기 위해 완료되어야 하는 작업 목록
4. 이전의 작업들이 완료되었는지

유저의 요청마다 SagaCoordinator 가 동작하게 되고 유저의 요청마다 위의 값들은 고유해야 하기 때문에, 요청에 대한 SagaState 가 생성되어야 했습니다.

Collection, Wrapper 로 위의 정보와 로직을 구현

단순히 이를 Collection 과 Wrapper 클래스로 만들어 SagaState 에서 관리하게 된다면, 각 응답을 받아 작업의 상태를 단순히 업데이트하는 SagaState 클래스에서는 위의 정보를 바탕으로 이전 작업의 의존성 관리와 성공 여부 추적 등 추가적인 로직이 포함되어 단일 책임 원칙 등 객체지향 원칙을 위배할 수 있는 구조가 될 것이라 생각하였습니다.

Netty 의 Promise 로 구현

Netty Promise 를 사용하게 된다면 addListener 로 다음 작업을 등록하여 실행 여부 판단과 성공으로 인한 다음 작업 실행등 로직을 분리할 수 있으며, 또한 PromiseAggregator 를 사용하면 이전에 의존하는 작업이 여러개인 경우에도 쉽게 처리할 수 있습니다.

Orchestration Saga 의 설정 예시는 다음과 같습니다.

SagaDefinition<Integer> sagaDefinition = sagaDefinitionFactory
        .addStep("step1", "{desination - 토픽 이름}", "{operation - Function(id, previous result, return)}", messageProducer, StepType.EXECUTE)
        .addStep("step2", "none", operation, messageProducer, StepType.EXECUTE)
        .addStep("step3", "none", operation, messageProducer, StepType.EXECUTE)
        .addStep("step4", "none", aggregateOperation, messageProducer, StepType.EXECUTE, "step2", "step3")
        .addStep("step5", "none", operation, messageProducer, StepType.EXECUTE, "step4")
        .addStep("step5", "none", operation, messageProducer, StepType.COMPENSATE)
        .addStep("step6", "none", aggregateOperation, messageProducer, StepType.EXECUTE, "step1", "step2", "step5")
        .addStep("step7", "none", operation, messageProducer, StepType.EXECUTE, "step6")
        .getSagaDefinition();

위의 설정은 아래와 같은 실행 체인을 구성하도록 하여, 이전 작업이 끝나면 자동적으로 다음 작업이 시작되도록 구현하였습니다.

 1, 2, 5 번 SagaPromise 가 완료되면 6번이 execute 된다.
 2, 3 번 SagaPromise 가 완료되면 4번이 execute 된다.

           +-> [1] --+
           |         | ------------> [6] --------> [7]
           |    +----+                ^
           |    |                     |
  [0] ---> +-> [2]                    |
           |    |                     |
           |    +----+                |
           |         | ---> [4] ---> [5 (Compensable)]
           +-> [3] --+

 [2, 3] done
  |
  v
 [4] --------->+ request -------+
               |    Service A   |
 [4] <---------+ response ------+
  |
  v
 [5] ----------> request Service B...

관련 이슈

11

추가 정보

• [x] 테스트 코드
• [ ] SagaPromise 가 모두 성공했을때 서비스에서 이를 처리해야 하는지, 서비스와 통합하면서 구현해야함 - #18