안전한 이벤트 발행 방법

0. 시작

이벤트는 강결합을 제거하거나 데이터 동기화 등 여러 유용한 상황에서 사용할 수 있다.

 

하지만, 언제나 예외가 발생할 수 있기 때문에 단순하게 이벤트를 발행/구독을 구현하는 것만으로 안전하지 않다.

 

여기서 "안전하다"는 여러 조건이 있을 수 있겠다.

 

• 메세지 발행이 보장되는지
• 발행된 메세지가 올바르게 전달되었는지
• 전달된 메세지가 올바르게 처리되었는지
• 중복 메세지가 중복 처리되지 않았는지
• (필요한 경우) 메세지의 순서가 보장되었는지

 

이벤트 발행/구독에서 사소한 오류 하나가 심각한 영향을 불러올 수 있기 때문에 안전함은 매우 중요하다.

 

오늘은 이벤트 발행 과정에서 고려해야할 여러 예외 상황을 알아보고, 그에 대한 대처를 알아보도록 하겠다.

 

1. 예외

먼저, 발생할 수 있는 예외들을 살펴보겠다.

 

네트워크 관련 예외 - 브로커와의 통신에서 오류가 발생하는 경우

이벤트를 관리하는 브로커로와의 통신에서 예외가 발생할 수 있다.

Kafka에서는 아래와 같은 예외들이 있다.

• TimeoutException: 브로커 응답 대기 시간 초과
• SocketTimeoutException: 소켓 연결 시간 초과
• ConnectionException: 브로커 연결 실패
• NetworkException: 일반적인 네트워크 오류

한 번 브로커와의 통신이 완료되었다고 안심하면 안 된다. 언제든지 브로커는 다운될 수 있다는 생각을 가지고 구현해야 한다.

 

 

 

 

메세지 직렬화/역직렬화 예외

Kafka에서는 미리 직렬화/역직렬화를 위한 클래스를 설정해둔다.

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
// ...
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

 

만약 설정과 맞지 않는 이벤트가 발행되면 예외를 일으킬 것이다.

 

이는 Kafka가 아닌 다른 프레임워크에서도 일어날 수 있다.

 

 

 

브로커 관련 예외

브로커가 항상 기대하는 메세지를 받는다고 생각하면 안 된다고 생각한다.

 

아래는 브로커에서 일어날 수 있는 여러 예외들이다.

 

• 토픽 이름이 잘못됨
• 토픽 접근 권한이 없음
• 복제본(Replica) 부족
• 리더 노드(Leader Node) 사용 불가

 

위에 대한 오류들을 애플리케이션 단에서 감지하여 처리해야 할 것이다.

 

 

비즈니스 로직 예외

단순히 이벤트 프레임워크와 상관없이, 메세지를 받고 처리하는 과정에서 예외가 발생할 수 있다.

 

이벤트 종류, 프레임워크 등에 따라서 여러 예외가 발생할 수 있고, 그에 대한 대처도 다르다.

 

• 유효성 검증 실패
• 이벤트 중복
• 리소스 없음

 

 

애플리케이션 종료

사용자의 요청을 처리하는 과정에서 갑작스러운 애플리케이션 종료 또한 고려해야할 사안이다.

 

이벤트 발행 전/ 후 모두 고려해야 할 것이다.

 

 

이벤트 순서 보장

서비스에 따라서 발행되는 이벤트의 순서에 따라 처리해야할 수도 있다. 

 

상황에 따라서 이전 이벤트가 완료된 후 실행해야 할 수도 있고,

 

프레임워크에 따라서 이벤트 순서가 보장되지 않는 경우도 있기 때문에 이를 고려해야 할 수도 있다.

 

 

이벤트 버전 확인

상황에 따라 이벤트의 버전이 바뀌는 시점도 고려해야 한다.

@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS)
public interface Event {
    String getVersion();
    String getType();
}

 

위와 같은 인터페이스를 통해 이벤트 버저닝을 하고, 각 버전에 따라 이벤트 프로세서를 바꿔야 할 수도 있다.

 

 

2. 해결

그렇다면, 위의 예외들을 해결할 수 있는 방법들이 무엇이 있을지 알아보겠다.

 

2-1. 트랜잭션 동기화

이벤트 발행의 중요한 점은 이벤트 발행 함수와 이벤트 실행 함수의 트랜잭션이 분리되어야 한다는 것이다.

 

만약 다음과 같이 주문을 하면 주문이 완료되었다는 메세지를 보내는 과정이 있다고 해보자.

 

만약 이 과정이 한 트랜잭션에 묶여있다면, 여기서 주문 완료 메세지를 보내는 과정이 실패하면 어떻게 될까?

 

 

주문 완료 메세지에서 발생한 예외가 호출 스택을 따라가서 주문에도 영향을 끼쳐서, 주문도 실패하는 상황이 벌어질 것이다.

 

따라서, 트랜잭션이 모두 완료된 후에 이벤트 발행을 동기화 해야한다.

 

 

주문 완료 메세지 전송이 실패하더라도 주문은 완료되어야 한다.

 

 

모놀리식에서 사용하는 Spring Application Event에서는 @TranscationalEventListener를 사용할 수 있다.

public class OrderEventHandler {
    private final EmailService emailService;

    // 트랜잭션 커밋 후 실행
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info("Order created: {}", event.getOrderId());
        emailService.sendOrderConfirmation(event.getOrderId());
    }
    
    // 트랜잭션 롤백 시 실행
    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void handleOrderFailed(OrderCreatedEvent event) {
        log.error("Order creation failed: {}", event.getOrderId());
        // 실패 처리 로직
    }
}

 

각 phase에 따라 트랜잭션 성공/ 실패 후 동작을 나눌 수 있다.

 

 

 

2-2. TransactionalOutbox Pattern

 

TransactionalOutbox 패턴은 발행한 메세지를 브로커에 전송하면서 DB에 저장해두는 방법이다.

 

Outbox는 "보낸 편지함"을 의미한다. 

 

https://microservices.io/patterns/data/transactional-outbox.html

 

 

Transactional Outbox Pattern은 보통 크게 Polling & Publisher 방법과 Transaction Log Tailing 방법으로 나뉜다.

 

2-2-1. Poliing & Publisher

Polling은 일정한 주기를 가지고 서버와 응답을 주고받는 방법을 말한다.

 

발행할 메세지를 Outbox DB에 저장해두고, 주기적으로 polling하여 이벤트를 발행하는 방법이다.

 

해당 방법으로 이벤트 발행을 보장할 수 있다.

 

아래 예시 구현으로 살펴보겠다.

 

@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    private String aggregateType;    // "ORDER", "PAYMENT" 등
    private String aggregateId;      // 해당 도메인 객체의 ID
    private String eventType;        // "ORDER_CREATED", "PAYMENT_COMPLETED" 등
    private String payload;          // 이벤트 데이터 (JSON)
    
    @Enumerated(EnumType.STRING)
    private EventStatus status = EventStatus.PENDING;
    
    private int retryCount = 0;
    private LocalDateTime createdAt;
    private LocalDateTime processedAt;
    
    @Version
    private Long version;            // 낙관적 락을 위한 버전
}

 

위와 같이 Outbox DB에 저장할 이벤트 객체를 정의한다.

 

public void createOrder(OrderRequest request) {
    //주문 생성
    Order order = orderRepository.save(Order.from(request));

    //Outbox 이벤트 저장
    OutboxEvent event = OutboxEvent.builder()
    	//...
        .build();

    outboxRepository.save(event);
}

 

주문을 생성하면 Outbox DB에 이벤트를 저장한다.

 

@Component
@EnableScheduling
@Slf4j
public class OutboxEventPublisher {
    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    
    @Scheduled(fixedRate = 1000) // 1초마다 실행
    @Transactional
    public void publishEvents() {
        List<OutboxEvent> events = outboxRepository
            .findByStatusAndRetryCountLessThan(
                EventStatus.PENDING, 
                3,
                PageRequest.of(0, 100)
            );
            
        for (OutboxEvent event : events) {
            try {
                // Kafka로 이벤트 발행
                kafkaTemplate.send(
                    event.getAggregateType().toLowerCase(), 
                    event.getAggregateId(),
                    event.getPayload()
                ).get(10, TimeUnit.SECONDS); // 타임아웃 설정
                
                // 발행 성공 시 상태 업데이트
                event.setStatus(EventStatus.PUBLISHED);
                event.setProcessedAt(LocalDateTime.now());
                outboxRepository.save(event);
                
            } catch (Exception e) {
                log.error("Failed to publish event: {}", event.getId(), e);
                
                // 재시도 횟수 증가
                event.incrementRetryCount();
                
                // 최대 재시도 횟수 초과 시
                if (event.getRetryCount() >= 3) {
                    event.setStatus(EventStatus.FAILED);
                }
                
                outboxRepository.save(event);
            }
        }
    }
}

 

이제 일정 주기마다 OutboxDB로부터 메세지를 Polling해서 이벤트를 발행한다.

 

이벤트의 발행이 완료되면 status를 바꾸는 방식으로 구현하거나, 대기 메세지/완료 메세지 테이블로 나누어 쿼리 과정에서 Join으로 사용해도 괜찮을 것 같다.

 

@Component
@EnableScheduling
public class FailedEventProcessor {
    @Scheduled(cron = "0 0 * * * *") // 매시간 실행
    public void processFailedEvents() {
        List<OutboxEvent> failedEvents = outboxRepository
            .findByStatus(EventStatus.FAILED);
            
        // 실패한 이벤트 처리 (알림, 로깅 등)
    }
}

 

상태에 따라 실패한 이벤트를 따로 처리할 수도 있다.

 

여기서 해결해야 할 문제는 여러개 더 있다.

• 인스턴스당 장애 발생 시 실행할 여러 노드를 배치
  • 복수 노드로 사용할 시 동시성 이슈 -> 락킹을 통해 해결
• 버저닝 -> 락킹 사용

위 방법들은 다음에 더 자세하게 다루는 기회에서 알아보도록 하겠다.

 

이로써 주문 트랜잭션 후에 메세지 publish를 보장할 수 있다.

 

해당 방법은 비교적 구현이 간단하고 편리하지만, DB로부터 주기적으로 Polling하기 때문에 DB에 부하가 갈 수 있다는 점이 단점이다.

 

 

 

2-2-2. Transaction Log Tailing 방법

 

Transaction Log Tailing, https://microservices.io/patterns/data/transaction-log-tailing.html

 

 

Transaction Log Trailing은 DB 트랜잭션의 로그를 사용하는 방법이다.

 

DB에 트랜잭션이 처리되면 생성되는 로그에 대한 CDC를 구현하는 방법이다.

 

CDC(Change Data Capture)
데이터 변경을 추적하여 변경 사항을 실시간으로 전달하는 데이터 통합 기술

 

데이터베이스에 따라 여러 로그를 활용할 수 있다.

• MySQL binlog
• Postgres WAL
• AWS DynamoDB table streams

위 데이터베이스 로그를 활용하는 방법은 구현 난이도가 높아 보통 툴을 사용하는데, 대표적인 툴은 아래와 같다.

 

• 디비지움(Debezium)
• 링크드인 데이터 버스(LinkdIn Databus)
• DynamoDB 스트림즈
• 이벤추에이트 트램

 

@Component
public class MySQLBinLogTailer {
    private final BinaryLogClient client;
    
    public MySQLBinLogTailer() {
        client = new BinaryLogClient(
            "localhost", 
            3306, 
            "username", 
            "password"
        );
        
        client.registerEventListener(event -> {
            if (event.getData() instanceof WriteRowsEventData) {
                WriteRowsEventData writeData = (WriteRowsEventData) event.getData();
                // 새로운 Row 데이터 처리
                processNewRows(writeData);
            }
        });
    }
    
    @PostConstruct
    public void start() {
        new Thread(() -> {
            try {
                client.connect();
            } catch (IOException e) {
                log.error("Failed to connect to MySQL binlog", e);
            }
        }).start();
    }
}

 

위와 같이 Mysql binlog를 활용하거나,

 

{
    "name": "mysql-outbox-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "localhost",
        "database.port": "3306",
        "database.user": "debezium",
        "database.password": "dbz",
        "database.server.name": "mysql",
        "database.server.id": "1",
        "database.include.list": "mydb",
        "table.include.list": "mydb.outbox_events",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.outbox",
        
        "transforms": "outbox",
        "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
        "transforms.outbox.table.field.event.id": "id",
        "transforms.outbox.table.field.event.key": "aggregate_id",
        "transforms.outbox.table.field.event.type": "event_type",
        "transforms.outbox.table.field.event.payload": "payload",
        "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType,aggregate_type:header:aggregateType",
        "transforms.outbox.route.by.field": "aggregate_type",
        
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": false
    }
}

 

위와 같이 Debezium을 설정해두고 사용하는 방법이 있겠다.

 

 

Transaction Log Tailing 방법은 Polling & Publisher 방법에 비해 Polling 작업이 없다는 점에서 장점이 있지만,

Kafka 메세지 형식에 맞추어 다시 데이터 dto를 빌드해야 한다.

 

 

2-3. Saga 패턴 : 트랜잭션 관리

 

Saga 패턴은 애플리케이션 내 트랜잭션을 순차적으로 처리하는 패턴이다.

여기서 트랜잭션은 DB가 아닌 애플리케이션 단의 트랜잭션 범위를 말한다.

https://learn.microsoft.com/ko-kr/azure/architecture/reference-architectures/saga/saga

 

핵심은 보상 트랜잭션이 있다는 것이다.

 

보상 트랜잭션
반대 효과를 가진 트랜잭션을 처리하여 잠재적으로 되돌릴 수 있는 트랜잭션
즉, 이전에 완료된 트랜잭션을 롤백하기 위해 역순으로 실행되는 트랜잭션

 

Saga패턴은 보통 MSA 환경에서 많이 등장하는데,

모놀리식과 달리 DB가 분산되어 있어 ACID가 지켜지는 트랜잭션을 사용할 수 없기 때문이다.

 

이를 예전에는 2PC(2-Phase-Commit)으로 해결했지만, 이는 상당히 비효율적이기 때문에 Saga 패턴이 해결점으로 등장했다.

 

 

 

Saga 패턴은 크게 코레오그래피(Choreography)와 오케스트레이션(Orchestration)이 있다.

 

Choreography

 

코레오그래피는 메세지 브로커를 이용해서 중앙 집중식 제어 지점 없이 이벤트를 전송하는 방법이다.

 

 

각 서비스의 로컬 트랜잭션이 완료되면 이벤트를 발행하고, 다음 수행해야할 앱이 받아서 다음 로컬 트랜잭션을 실행하는 방식으로 사용된다.

 

 

만약 중간에 트랜잭션이 실패하면 실패한 앱에서 보상 이벤트를 발행하여 롤백을 위한 보상 트랜잭션을 각 앱에 실시하도록 구성한다.

 

 

해당 방법은 유지/관리가 필요하지 않고 구현이 간단하지만 Saga 참가자가 어떤 명령을 수신 대기하는지 추적하기 어렵고, 트랜잭션의 상태를 확인하기 어렵다는 단점이 있다.

 

Orchestration

 

 

오케스트레이션은 중앙 집중식 컨트롤러(Orchestrator)가 실행할 로컬 트랜잭션을 알려주는 방법이다.

 

오케스트레이터는 모든 트랜잭션을 처리하고, 이벤트에 따라 수행할 작업을 각 참가자한테 알려준다.

 

각 참가자는 요청을 실행하고, 실패 시 보상 트랜잭션을 사용하여 오류 복구를 시도한다.

 

 

 

2-4. 재전송 : 일시적 문제 해결

 

이벤트는 어떤 이유에서라도 전송을 실패할 수 있다.

 

전송에 실패한 이벤트를 재시도 하는것은 자연스러운 수순인데,

 

이는 어느정도 일시적인 문제를 해결할 순 있겠지만, 영구적인 문제는 해결할 수 없다.

 

재시도 정책은 기본적으로 Spring Retry를 이용해서 구현할 수 있다.

 

Spring Retry를 사용하려면 @EnableRetry를 설정해야 한다.

 

@Retryable(
    value = {KafkaException.class, TimeoutException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000, multiplier = 2)
)
public void publishEvent(String topic, String key, Object event) {
    // 메세지 전송 로직
}

@Recover
public void handlePublishFailure(Exception e, String topic, String key, Object event) {
    // 실패 처리 로직
}

 

최대 전송 횟수는 maxAttempts, 재시도 메세지간 간격은 backoff로 설정할 수 있다.

 

실패한 메세지에 대한 처리는 @Recover를 통해서 처리할 수 있다.

 

 

 

 

 

2-5. DLT (Dead Letter Topic) : 실패한 메세지 관리하기

DLQ는 AWS SQS에서 사용하는 소비되지 못한 메세지(Dead Letter)를 담아두는 대기열(큐)이다.

 

https://aws.amazon.com/ko/blogs/compute/using-amazon-sqs-dead-letter-queues-to-replay-messages/

 

실패된 메세지를 Lambda를 통해 어떤 처리를 할지 정해둘 수 있다.

 

Kafka에서는 DLQ가 아닌 DLT(Dead Letter Topic)을 사용한다.

 

factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

 

Kafka 메세지 전송이 실패하는 경우를 확인하려면 수동 ACK모드를 사용하는 설정이 권장된다.

 

@RetryableTopic
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

@DltHandler
public void processMessage(MyPojo message) {
    // ... message processing, persistence, etc
}

 

메서드가 같은 형태의 @RetryableTopic과 @DltHandler를 붙인 메서드로 DLT를 활용할 수 있다.

 

또는, RetryTopicConfiguration을 통해 설정해둘 수 있다.

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltHandlerMethod("myCustomDltProcessor", "processDltMessage")
            .create(template);
}

@Component
public class MyCustomDltProcessor {

    private final MyDependency myDependency;

    public MyCustomDltProcessor(MyDependency myDependency) {
        this.myDependency = myDependency;
    }

    public void processDltMessage(MyPojo message) {
        // ... message processing, persistence, etc
    }
}

 

@RetryableTopic에 여러 애트리뷰트를 달아놓으면 코드가 중복되고 복잡해질 수 있다.

 

RetryTopicConfiguration을 사용하면 전역적으로 재시도 토픽과 DLT 토픽이 적용되므로 더 깔끔한 적용인 것 같다.

 

여기서 중요한 점은, @KafkaListener의 Factory 설정과 동일해야 MessageConversionException을 피할 수 있다.

 

 

 

 

참고
https://ridicorp.com/story/transactional-outbox-pattern-ridi/
https://giron.tistory.com/156
https://docs.spring.io/spring-kafka/reference/index.html
https://docs.spring.io/spring-kafka/reference/kafka/micrometer.html
https://docs.spring.io/spring-kafka/reference/kafka/annotation-error-handling.html
https://hvho.github.io/2021-12-05/spring-retry
https://junuuu.tistory.com/795
https://www.kai-waehner.de/blog/2022/05/30/error-handling-via-dead-letter-queue-in-apache-kafka/
https://docs.spring.io/spring-kafka/reference/retrytopic/dlt-strategies.html
https://velog.io/@wwlee94/Kafka-%EC%9E%AC%EC%8B%9C%EB%8F%84-DLT-%EB%B9%8C%EB%8D%94-%EC%A0%91%EA%B7%BC-%EB%B0%A9%EC%8B%9D%EC%9C%BC%EB%A1%9C-%EB%A6%AC%ED%8C%A9%ED%86%A0%EB%A7%81