1. 개요
지금까지 구성한 Event 도메인 객체나 EventListener 등은 Message를 Broker로부터 전달받아 이를 처리하는 Consumer에 대한 내용들이었다.
이제 Producer를 어떻게 구성할 것인지 그 전략에 대해 구상해볼 차례이다.
Producer 역시 마찬가지로 단순히 Message를 보내는 것에 그치지 않고, RDB(MySQL)와 Kafka 간 이루어지는 분산 트랜잭션의 원자성을 어떻게 보장할 수 있으며, 최종적으로 트랜잭션을 완료한 이후 데이터 일관성까지 어떻게 확보할 것인지 고민해야 한다.
이에 대한 전략을 구상하기 위해 찬찬히 살펴보고 생각한 과정을 기록한다.
2. 문제상황인지
일단 지금의 상황에서 어떤 점을 고려해야 하며, 어떤 점이 문제인지 찬찬히 살펴보도록 하겠다.
2-1. 문제상황 #1 - 분산 시스템에서의 분산 트랜잭션
Producer 입장에서 처리해야하는 트랜잭션은 크게 두가지로 볼 수 있겠다.
- 요청한 서비스 처리 - MySQL
- 해당 서비스에 대한 Kafka 메시지 전달을 위해 이벤트 전송
즉 하나의 트랜잭션에서 처리해야 하는 각각의 처리는 서로 다른 시스템에서 발생하므로(일어나므로), 이 분산 시스템에서의 분산 트랜잭션의 원자성과 데이터 일관성을 모두 확보하기 위한 방안을 고민해야 한다.
2-2. 문제상황 #2 - Kafka 측의 장애 발생 및 이로 인한 장애전파
또 다른 문제상황으로 Kafka에서 장애가 발생한 경우를 생각해 볼 수 있다.
Producer에서 전송한 이벤트를 Consumer가 받아 처리하기 위해선 네트워크를 통한 복잡한 처리과정이 필요(로직의 복잡성도 있지만, Consumer로 메시지를 전달하기까지의 과정적 복잡성도 포함)하다.
이때 이벤트 전송 과정에서 발생하는 네트워크 순단 및 장애로 인해, Producer 이벤트 전송 불가 및 지연, Consumer의 무한대기 등 각 항목에서 예외없이 장애전파 및 이로 인한 데이터 유실 위험이 존재한다.
3. 전략구상
위 문제상황을 바탕으로 Producer 구상방안을 어떻게 진행하면 좋을지 찬찬히 생각해보았다.
3-1. 전략구상 #1 - 기준을 잡는다.
먼저, 시스템이나 특정 처리 과정 등, 어떤 항목을 최대한 신뢰할 수 있겠는가, Producer 전략을 구성하기 위한 기준점을 먼저 생각해본다.
나의 경우엔 Kafka 체계를 기준으로 잡아보았다.
- Kafka 자체는 동일 topic에서 여러 partition을 두어 메시지 분산 저장 및 병렬 저장으로 데이터를 정상적으로 수신받았다면 정상복구가 가능한 체계이다.
따라서 이 아이디어를 바탕으로, Consumer 측에서 이벤트 처리 및 성공응답(commit) 송신 시 데이터 유실없이 처리가 가능한 점을 기준으로 보았다.
이에 따라 Kafka/Consumer 측에서 발생하는 데이터 유실보다, 애초에 Producer에서 발생하는 장애상황(Message 전송 후 네트워크 순단 등..)에 대해 대비하여 그에 중점적인 전략을 구상하는 것이 나을 것으로 생각하였다.
3-2. 전략구상 #2 - Producer의 정상 동작 관점
Producer의 장애상황이 문제라면, 첫번째 장애상황으로 Producer의 정상 동작에 문제가 생겼을때를 생각해보았다.
Producer -> Kafka로 메시지 전송과는 무관하게 Producer의 동작은 정상적으로 이루어져야 한다.
즉, Kafka의 데이터 전송이 실패한다고 하여 Producer의 동작 자체도 멈추면 안되는데, 사실 Kafka 이벤트 전송은 Producer의 전송 이후에 일어나는 것이기도 하고 애초에 Kafka의 장애로 인해 Producer의 동작까지 장애가 번진다는 상황이 그리 고려할만한 상황은 아니라는 생각이 들어 다음의 관점을 생각하였다.
3-3. 전략구상 #3 - Producer의 정상 동작 이후의 이벤트 메시지 전송 관점
Producer는 항상 정상적으로 동작하며, 애초에 이벤트 메시지를 전송하기 위해선 Producer의 동작이 선행되어야 한다.
Producer의 정상 동작 이후에 Producer가 생산 및 전파하는 이벤트 데이터까지 유실하지 않고 정상적으로 이루어져야 한다.
최종적인 문제상황은 여기서 발생하며, Producer 전략을 구상하기 위한 포인트는 바로 여기에 있다.
Producer에서 이벤트 메시지 전송을 요청하였을때 그 과정에서 장애가 발생한다면, Consumer 측에서 해당 메시지를 이어받지 못해 데이터 일관성이 깨지고 그만큼의 유실이 발생한다.
3-4. 전략구상 #4 - 결론
Eventuall Consistency
따라서, 위 문제상황 및 전략구상 과정을 바탕으로 서비스 비즈니스 로직 수행과 이벤트 전송은 어떠한 경우에도 반드시 단일 트랜잭션으로 이루어져야 한다는 것을 알 수 있다.
다만 이 보장은 반드시 실시간으로 이루어질 필요는 없고, 최종적인 일관성만 유지한다면 무방하기에 비동기 처리까지 전략구상에 고민해 보면 좋겠다.
4. 최종방안 구상하기
이제 위의 문제상황과 전략구상 과정을 모두 종합하여, 분산 트랜잭션을 어떻게 구상해야 할지 최종방안을 구상해볼 차례이다.
4-1. 단순하게 이벤트 전송을 추가하여 단일 트랜잭션화 할 수 없다.
말 그대로, 이벤트 전송에 대한 서비스를 구성하여 이를 기존 서비스에 단순하게 추가하는 것만으로는 분산 트랜잭션의 원자성과 데이터 일관성을 보장할 수 없다.
이에 대한 내용을 위와 같이 정리해보았는데,
1) 비즈니스 로직만 있는 상황(비즈 only)의 경우, 단순하게 Transaction start 후 commit, rollback까지 Spring Framework가 단일 스레드를 활용하여 전체적인 트랜잭션의 원자성을 보장해줄 수 있다(Transactional 어노테이션 활용할 경우).
2) 하지만 이 비즈니스 로직 Transaction에 이벤트 전송 과정을 단순하게 붙인다면, Kafka 장애발생으로 인해 해당 처리 지연 시 전체 트랜잭션 지연 및 이로 인한 무한대기 등의 치명적인 하자점이 발생할 수 있다.
또한 지금과 같이, 기본적으로, MySQL의 상태변경(RDB)과 Kafka 이벤트 전송(Kafka)을 묶지 못하고 이를 위한 기능 자체를 제공하지 않는다.
위 그림처럼 publishEvent를 중간에 넣어 FacadeService화 할 수 있겠지만, Kafka 장애로 인해 해당 과정이 블로킹되거나 작업지연이 발생할 경우 전체 트랜잭션이 완료되지 못하는 장애 전파(Kafka 장애 > application 장애)가 발생할 수 있겠다.
또한 반대로 트랜잭션이 Commit되었으나 이벤트 전송이 실패하여 데이터 일관성이 깨지는 복잡한 문제도 발생할 수 있다.
3) publish event만 비동기 처리한다면, 비동기 처리 이벤트 전송이 실패할 경우 이에 대한 MySQL의 보상 트랜잭션을 만들어 rollback하거나 후처리 등을 해야 한다. 보상트랜잭션 자체가 복잡한 과정이기도 하고, 보상 이후의 정상적인 rollback 등의 동작이 일어날 것이란 보장을 확정할 수 없겠다.
4-2. 최종방안 - Transactional Messaging
이러한 문제점을 개선할 수 있는 분산 트랜잭션 방안을 생각해보았다.
최종적으로는, 분산 시스템에서 단일 트랜잭션의 원자성을 확보하고, 다만 실시간으로 처리하지는 않고 비동기로 데이터 일관성을 보장할 수 있는 처리방안을 마련하여 Transactional Messaging을 통해 최종적인 전략 구상을 결론지었다.
5. 이론적 Transactional Messaging 구현 방안
이론적으로 Transactional Messaging을 구현할 수 있는 방안은 크게 3가지가 존재한다.
이에 대해 먼저 공부해보았다.
5-1. Two Phase Commit
Two Phase Commit은 분산 시스템에서, 각 독립적인 시스템들의 트랜잭션 수행 이후 모든 시스템의 성공적인 처리 응답을 전달받아 중개자(Coordinator)가 최종 Commit 완료하는 분산 트랜잭션의 방안이다(실패 시 모두 rollback).
위와 같은 과정으로 이루어지는데, 중간의 Coordinator가 prepared phase와 commit phase를 두어 트랜잭션을 수행한다.
- Coordinator가 각 참여자에게 트랜잭션 수행 후 완료 준비(commit) 되었는지 질의하며, 각 참여자는 commit 준비 완료 응답을 보낸다.
- 모든 참여자가 준비완료 송신 후 Coordinator가 이를 수신 시, Coordinator는 모든 참여자에게 트랜잭션 commit을 요청, 최종적으로 모든 참여자가 commit 완료하여 트랜잭션을 종료한다.
하지만 모든 참여자의 응답을 기다려야 하기에 그만큼의 지연 발생, Coordinator나 참여자에 장애 발생 시 참여자들은 다른 상태를 모른채 무한 대기를 할 수 있다.
더불어 트랜잭션 복구 처리가 어렵기도 하고, 성능/지원적인 측면에서도 부족한 점이 많기에 해당 방법은 고려하지 않기로 하였다.
5-2. Transactional Outbox Pattern
이후 많이 들어보았을 Transactional Outbox pattern에 대해 고려해보았다.
실무에서도 몇번 사용한 pattern이기도 하고, 무엇보다 기본적으로 일반적인 트랜잭션에 이벤트 전송 작업을 포함시킬 수 없기에 이에 대한 정보를 별도로 관리할 수 있는 방안이 필요하다.
이처럼 이벤트 전송 정보를 DB트랜잭션에 포함하고 이를 기록하여, 분산 트랜잭션을 진행하도록 하고 해당 트랜잭션의 RDB에 outbox 테이블을 지정하여 서비스 로직과 outbox 이벤트 기록을 단일 트랜잭션으로 일단 묶는 방안을 생각하게 되었다.
5-2-1. 이벤트 기록과 전송은 따로 진행
다만 이전에 기술한대로 단일 트랜잭션은 MySQL에서 그대로 이행하되, 메시지를 전송하는 과정은 굳이 실시간으로 처리하지 않고 데이터 일관성을 보장할 수 있을 수준으로 비동기 처리(후처리)하도록 구성하면 되겠다.
따라서, 이 트랜잭션은 크게 두가지 phase로 나누어 구현하는 전략을 생각하였다.
- 이벤트 기록 ) 비즈니스 로직 수행 및 outbox 테이블에 해당 이벤트 정보를 기록
- Kafka 전송 ) outbox 테이블 미전송 상태를 조회하여 미전송 항목을 Kafka로 전송 및 전송완료처리
다만 중간에서 미전송 데이터를 읽고 Kafka로 보내는 Message Relay에 대한 고민이 필요하다.
5-3. Transaction Log Tailing
위에서 Message Relay를 통해 이벤트 메시지 정보를 읽고 이를 Broker에 전송하는 과정을 구현하지 않고도 RDBMS에서 제공하는 기능을 활용하여 상태변경을 추적할 수 있는 전략이 있기도 하였다.
Transaction Log Tailing은 DB의 트랜잭션 로그를 추적하고 분석하는 방법으로, DB는 각 트랜잭션의 변경사항을 로그로 기록한다.
- MySQL : binlog, PostgreSQL : WAL, SQL Server : Transaction log.
이러한 로그를 읽어서 Message Broker에 이벤트를 전송할 수는 있는데(=데이터 변경사항을 추적하여 다른 시스템에 해당 내용을 전송 가능), 이를 위해선 RDBMS 자체적으로 제공하는 CDC(Change Data Capture) 기술에 대한 이해가 필요하다.
위와 같이 Transaction log Miner를 통해 RDBMS에서 남기는 Transaction log를 조회하고 추적할 수 있으며, log 기반으로 Kafka Broker에 해당 이벤트를 전송할 수 있다.
이 경우 Outbox Pattern에서 outbox table은 필요가 없어지기에, 운영비용 관점에서 유리할 수 있다. 다만, 별도의 학습과 이에 수반하는 운영비용이 발생하여 해당 구현방안은 제외하기로 하였다(그리고 무엇보다 Outbox Pattern에서 활용하는 테이블 추가에 소모하는 비용이 유지관리 측면에서 제외해야할 정도로 큰 비용이 아닐 것이라는 판단을 하였다).
참고로, CDC 활용 시
- Data table의 변경사항을 관리하는 log 정보는 DB의 구조에 종속적이기에, log의 정보 및 구조적으로 한계가 있다.
5-4. 결론
따라서 익숙한 환경에서 Transactional Messaging을 구현하기에는 성능/구조/운용적인 측면에서 Outbox Pattern이 가장 적합할 것이라는 결론을 지었다.
6. Outbox Pattern 구현방안 구체화
이제 Outbox Pattern으로 어떻게 분산트랜잭션을 구현할 것인지 구체화하는 차례이다.
일단 구현방안을 정리해보면,
- 데이터 변경사항 및 outbox 이벤트 로그 기록은 단일 트랜잭션으로 관리한다.
- Message Relay를 통한 이벤트 기록 및 조회, 전송이 이루어질 수 있도록 한다.
6-1. outbox table 생성
outbox table은 outbox_id / event type / payload / shard_key / created_at 으로 Kafka에 전송할 이벤트 내용으로 구성해주었다.
Shard key 구성의 의미
이때 Shard Key의 경우, 현재 table들은 모두 샤딩을 한 체계임을 가정한 분산 환경이기에, 동일한 원리로 outbox table 역시 안전하게 빠른 이벤트 기록 및 전송이 가능하도록 분산 outbox table 환경을 가정하여 구성하도록 하였다.
또한 특정 도메인의 서비스처리 및 이벤트 기록에 대한 outbox pattern 처리도 같은 도메인에서 일어나도록 DB를 일치하여 구성하는 것이 좋겠다.
6-2. Message Relay 구현 전략
그리고 가장 중요한 Message Relay를 어떻게 구현할 것인지 그 전략을 구상해볼 차례이다.
게시글 서비스 요청을 예를 들어 Message Relay 구현 전략을 한번 구상해보았다.
최초 Article Service에서 API호출 발생 시, 단일 트랜잭션으로 구성한 게시글 서비스와 이벤트 메시지 outbox 서비스를 모두 진행하며 이는 모두 MySQL에서 이루어진다.
이때 Message Relay는 다음과 같은 정책으로 이벤트 메시지를 Kafka Broker에 전송한다.
- 단일 트랜잭션의 성공적인 처리를 위해 10초라는 여유있는 시간간격을 부여하며, Message Relay는 최초 서비스 요청 후 10초 이후에 메시지 미전송 혹은 실패 항목들을 조회한다.
- 이후 주기적으로, 10초 간격으로 미전송 및 전송실패 메시지를 조회하여 Kafka에 다시 전송한다.
하지만 이 10초간격의 polling 만으로는 그 사이에 쌓이는 이벤트 메시지를 처리하기엔, 즉 모든 이벤트 메시지를 비동기로 처리하기엔 DB부하 및 성능적인 관점에서 매우 불리할 수 있다.
이에 대한 단점을 보완하고자, 최초 서비스 요청 시에도 동시에 Message Relay에 이벤트 메시지를 직접 전송한다.
- 서비스 입장에서는 outbox 메시지 기록과 함께 메시지 전송 시 성공기록도 즉시적으로 기록할 수 있다.
- Message Relay 입장에서는 서비스로 요청한 raw 이벤트 메시지 전부 다 Broker에 전달하는게 아니라, 미전송 및 실패한 항목에 대해서만 Broker에 재전송하면 된다.
추가적으로, 10초 polling으로 인한 이벤트 메시지 전송이 힘들다면 게시글 서비스 및 댓글 트래픽 규모에 따라 간격을 낮추는 등 조정할 수 있다.
6-3. 이벤트 메시지 전송 상태값 관리
프로젝트 관리 컨벤션이나 이벤트 전송 의도에 따라 상태값 관리 방안이 달라질 수 있다.
- Relay가 이벤트 단순 전송 용도 -> 이벤트 전송 완료 후 삭제
- Relay가 이벤트 추적 및 리플레이 필요 시(Event Sourcing) -> 이벤트 전송 완료 상태 변경
추가적인 Consumer 데이터 일관성 및 멱등성 보장방안도 생각해볼 수 있겠지만, 지금으로서는 위 방안이 가장 현실적이고 효율적이라는 생각이 들었다.
7. 적절한 Polling 전략 - Message Relay 중앙저장소 및 Coordinator 배치하기
이제 마지막으로, 적절한 방식의 Polling을 구현하기 위한 전략을 구상해본다.
분산시스템에서 Message Relay가 oubox 테이블로 polling할때, Article/View/Like 등 각 서비스에서 운용하는 모든 Outbox 테이블에 직접 polling해야하는 번거로움이 생긴다.
데이터나 메시지 이벤트에는 샤딩할 수 있는 논리적 기준인 shard key가 존재하지만, polling 행위 자체를 샤딩할 수 있는 기준(shard key)는 존재하지 않기 때문이다.
이로 인해, application 실행 및 polling할 경우 다른 application에서 동일한 outbox에 polling하여 메시지를 중복 처리할 수도 있고, 기본적으로 모든 샤딩 outbox 테이블에 polling 해야하기에 엄청난 성능적 불리함이 생긴다.
이를 위해, 각 application은 서비스 요청 시점에 활성화되어 본인이 특정 샤드를 할당받아 해당 샤드에 대해서만 polling하는 전략을 구상하기로 하였다.
즉, 메시지를 요청한 application 자신이, polling을 하는 샤드의 일부를 할당 받아 직접 Message Relay를 처리하며 이때 자신을 실행한 application의 식별자와 현재시간으로 일종의 기준(샤드키)을 지정받는다.
이 지정받은 식별자로 특정한 샤드에 대해서만 polling을 진행하여 모든 oubox 샤드를 polling하지 않아도 되고, 이에 따른 중복처리 위험도 감소할 수 있게 된다.
이때 Coordinator는 3초 간격으로 어느 샤드를 바라보아야할지, 그 식별자와 생성시간을 저장한 중앙저장소에 ping을 전송하여 실행중인 application의 목록을 파악하여 그 application을 본체로 outbox에 polling하여 이벤트 메시지를 처리하도록 중재한다(샤드키 적절하게 분산).
중앙저장소는 마지막 ping 수행 후 9초가 경과시 실행중인, 해당 application은 종료되었다고 판단하여 목록에서 제거하며 이 중앙저장소는 빠른 처리를 위해 Redis Sorted Set(app id - score(ping created time))을 활용한다.
8. 인덱스 생성
outbox에서 미전송 이벤트 메시지를 조회하기 위해 이에 대한 인덱스 생성도 필요하다.
그리 복잡한 생각 필요없이, shard key와 생성시간 순서대로 조회를 할 것이기에 이와 상응하는 index를 생성하면 되겠다.
