Spring Modulith 이벤트 외부화

글쓴이 /

Spring Modulith Event Externalization이란?

Event Externalization은 Spring Modulith 1.1에서 도입된 기능으로, 모듈 내부 도메인 이벤트를 Kafka, RabbitMQ, SQS 등 외부 메시지 브로커로 자동 발행하는 메커니즘입니다. 기존에 ApplicationEventPublisher로 발행한 이벤트를 코드 변경 없이 외부 시스템으로 내보낼 수 있어, 모듈러 모놀리스에서 마이크로서비스로의 점진적 전환을 지원합니다.

내부 이벤트 vs 외부화 이벤트

비교 항목 내부 이벤트 외부화 이벤트
전달 범위 같은 JVM 내 외부 브로커 + 다른 서비스
전달 보장 트랜잭션 커밋 후 동기/비동기 Event Publication Registry 기반
직렬화 Java 객체 그대로 JSON 자동 직렬화
라우팅 Spring 컨텍스트 내 토픽/큐 자동 결정
코드 변경 — 어노테이션 1개 추가

의존성 설정

브로커별로 필요한 의존성이 다릅니다.

// build.gradle.kts — Kafka 사용 시
dependencies {
    implementation("org.springframework.modulith:spring-modulith-starter-core")
    implementation("org.springframework.modulith:spring-modulith-events-kafka")
    implementation("org.springframework.modulith:spring-modulith-events-jpa") // Event Publication Registry
    
    // RabbitMQ 사용 시
    // implementation("org.springframework.modulith:spring-modulith-events-amqp")
    
    // AWS SQS/SNS 사용 시
    // implementation("org.springframework.modulith:spring-modulith-events-aws-sqs")
    // implementation("org.springframework.modulith:spring-modulith-events-aws-sns")
    
    // JMS 사용 시
    // implementation("org.springframework.modulith:spring-modulith-events-jms")
}

@Externalized: 이벤트 외부화 선언

도메인 이벤트 클래스에 @Externalized 어노테이션 하나로 외부 발행을 활성화합니다.

// 기본 사용: 클래스명이 토픽/라우팅 키가 됨
@Externalized
public record OrderCompleted(
    String orderId,
    String customerId,
    BigDecimal totalAmount,
    Instant completedAt
) {}

// 토픽과 라우팅 키 명시적 지정
@Externalized("order-events::#{#this.orderId()}")
public record OrderCompleted(
    String orderId,
    String customerId,
    BigDecimal totalAmount,
    Instant completedAt
) {}
// → Kafka 토픽: "order-events", 파티션 키: orderId 값

// SpEL로 동적 라우팅
@Externalized("#{#this.priority() == 'HIGH' ? 'urgent-orders' : 'order-events'}::#{#this.orderId()}")
public record OrderCompleted(
    String orderId,
    String customerId,
    BigDecimal totalAmount,
    String priority,
    Instant completedAt
) {}

// 발행 코드는 변경 없음 — 기존 내부 이벤트 발행 그대로
@Service
@Transactional
public class OrderService {
    private final ApplicationEventPublisher events;

    public void completeOrder(String orderId) {
        Order order = orderRepository.findById(orderId).orElseThrow();
        order.complete();
        orderRepository.save(order);

        // 내부 이벤트 발행 → @Externalized로 Kafka에도 자동 발행
        events.publishEvent(new OrderCompleted(
            order.getId(),
            order.getCustomerId(),
            order.getTotalAmount(),
            Instant.now()
        ));
    }
}

핵심 포인트: ApplicationEventPublisher.publishEvent() 코드는 전혀 변경하지 않습니다. 이벤트 클래스의 어노테이션만으로 외부 발행이 활성화됩니다.

Event Publication Registry: 발행 보장

외부화 이벤트의 최소 1회 전달(at-least-once)을 보장하는 핵심 메커니즘입니다.

// application.yml
spring:
  modulith:
    events:
      jdbc:
        schema-initialization:
          enabled: true  # event_publication 테이블 자동 생성
    republish-outstanding-events-on-restart: true  # 미발행 이벤트 재시도

// 동작 원리:
// 1. 트랜잭션 내에서 이벤트 발행 → event_publication 테이블에 INSERT
// 2. 트랜잭션 커밋 후 → 브로커로 전송 시도
// 3. 전송 성공 → event_publication에서 completion_date 업데이트
// 4. 전송 실패 → 재시작 시 또는 주기적으로 재시도

// 미완료 이벤트 재발행 스케줄링
@Configuration
public class EventPublicationConfig {

    @Bean
    public IncompleteEventPublications incompleteEvents(
            EventPublicationRepository repository) {
        return new IncompleteEventPublications(repository);
    }

    // 10분마다 미발행 이벤트 재시도
    @Scheduled(fixedRate = 600_000)
    public void resubmitIncompletePublications(
            IncompleteEventPublications publications) {
        publications.resubmitIncompletePublicationsOlderThan(
            Duration.ofMinutes(5));
    }
}
-- event_publication 테이블 구조
CREATE TABLE event_publication (
    id               UUID PRIMARY KEY,
    listener_id      TEXT NOT NULL,
    event_type       TEXT NOT NULL,
    serialized_event TEXT NOT NULL,
    publication_date TIMESTAMPTZ NOT NULL,
    completion_date  TIMESTAMPTZ  -- NULL이면 미완료
);

-- 운영 모니터링 쿼리
-- 미완료 이벤트 확인
SELECT event_type, COUNT(*), MIN(publication_date) AS oldest
FROM event_publication
WHERE completion_date IS NULL
GROUP BY event_type
ORDER BY oldest;

-- 오래된 미완료 이벤트 정리 (수동)
DELETE FROM event_publication
WHERE completion_date IS NULL
  AND publication_date < now() - INTERVAL '7 days';

커스텀 직렬화와 이벤트 변환

외부 발행 시 이벤트 페이로드를 변환하거나 필터링하는 방법입니다.

@Configuration
public class EventExternalizationConfig {

    @Bean
    public EventExternalizationConfiguration eventExternalizationConfiguration() {
        return EventExternalizationConfiguration.externalizing()
            // 특정 이벤트만 외부화
            .select(
                EventExternalizationConfiguration.annotatedAsExternalized()
            )
            // 라우팅 커스터마이징
            .route(
                OrderCompleted.class,
                it -> RoutingTarget.forTarget("order-events")
                    .andKey(it.orderId())
            )
            // 이벤트 페이로드 변환 (내부 필드 제거)
            .mapping(
                OrderCompleted.class,
                it -> new OrderCompletedExternal(
                    it.orderId(),
                    it.totalAmount(),
                    it.completedAt()
                    // customerId는 외부에 노출하지 않음
                )
            )
            // 조건부 외부화 (필터링)
            .select(
                OrderCompleted.class,
                it -> it.totalAmount().compareTo(BigDecimal.ZERO) > 0
            )
            .build();
    }
}

// 외부 발행용 축소된 이벤트
public record OrderCompletedExternal(
    String orderId,
    BigDecimal totalAmount,
    Instant completedAt
) {}

Kafka 설정과 토픽 매핑

Kafka를 브로커로 사용할 때의 상세 설정입니다.

// application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all                    # 모든 ISR 복제 확인
      retries: 3
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true   # 중복 발행 방지
        max.in.flight.requests.per.connection: 5

// 토픽 자동 생성 설정
@Configuration
public class KafkaTopicConfig {
    @Bean
    public NewTopic orderEventsTopic() {
        return TopicBuilder.name("order-events")
            .partitions(6)
            .replicas(3)
            .config(TopicConfig.RETENTION_MS_CONFIG, "604800000") // 7일
            .build();
    }

    @Bean
    public NewTopic urgentOrdersTopic() {
        return TopicBuilder.name("urgent-orders")
            .partitions(3)
            .replicas(3)
            .build();
    }
}

테스트: 외부화 이벤트 검증

Spring Modulith의 @ApplicationModuleTest와 Scenario API로 이벤트 외부화를 테스트합니다.

@ApplicationModuleTest
class OrderModuleIntegrationTest {

    @Test
    void shouldExternalizeOrderCompletedEvent(Scenario scenario) {
        // given
        var orderId = "ORD-001";

        // when + then
        scenario.stimulate(() -> orderService.completeOrder(orderId))
            .andWaitForEventOfType(OrderCompleted.class)
            .matchingMappedValue(OrderCompleted::orderId, orderId)
            .toArriveAndVerify(event -> {
                assertThat(event.orderId()).isEqualTo(orderId);
                assertThat(event.totalAmount()).isPositive();
            });
    }

    @Test
    void shouldPersistEventPublication(Scenario scenario) {
        // Event Publication Registry에 기록되는지 검증
        scenario.stimulate(() -> orderService.completeOrder("ORD-002"))
            .andWaitForEventOfType(OrderCompleted.class)
            .toArriveAndVerify(event -> {
                // event_publication 테이블에 completion_date가 설정되었는지
                var publications = jdbcTemplate.queryForList(
                    "SELECT * FROM event_publication " +
                    "WHERE event_type LIKE '%OrderCompleted%'");
                assertThat(publications).isNotEmpty();
                assertThat(publications.get(0).get("completion_date"))
                    .isNotNull();
            });
    }
}

// Testcontainers로 Kafka 통합 테스트
@SpringBootTest
@Testcontainers
class KafkaExternalizationTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired
    private KafkaConsumer<String, String> consumer;

    @Test
    void shouldPublishToKafkaTopic() {
        orderService.completeOrder("ORD-003");

        ConsumerRecords<String, String> records = 
            consumer.poll(Duration.ofSeconds(10));
        assertThat(records).hasSize(1);
        assertThat(records.iterator().next().topic())
            .isEqualTo("order-events");
    }
}

모놀리스→마이크로서비스 전환 전략

Event Externalization을 활용한 3단계 점진적 전환 전략입니다.

  1. 1단계 — 내부 이벤트 도입: 모듈 간 직접 호출을 ApplicationEventPublisher 기반 이벤트로 전환. 모듈 경계를 명확히 합니다.
  2. 2단계 — 외부화 활성화: 이벤트 클래스에 @Externalized 추가. 내부 리스너와 외부 브로커 동시 발행. 기존 코드 변경 제로.
  3. 3단계 — 모듈 분리: 외부 브로커를 통해 이벤트를 수신하는 독립 서비스로 모듈을 추출. 내부 리스너 제거, 외부 컨슈머로 전환.
// 2단계: 내부 + 외부 동시 운영
// 내부 리스너 (기존 유지)
@Component
public class InventoryEventListener {
    @TransactionalEventListener
    public void onOrderCompleted(OrderCompleted event) {
        inventoryService.deductStock(event.orderId());
    }
}

// 외부 컨슈머 (새 서비스에서 수신 준비)
// → 3단계에서 별도 서비스로 분리
@KafkaListener(topics = "order-events")
public void onOrderCompleted(OrderCompletedExternal event) {
    inventoryService.deductStock(event.orderId());
}

Spring Modulith Event Externalization은 Transactional Outbox 패턴을 프레임워크 레벨에서 자동 구현한 것입니다. 별도 Debezium CDC 파이프라인 없이도 이벤트 발행을 보장합니다. Spring Modulith 기본 개념은 Spring Modulith 모듈러 모놀리스 가이드를, Outbox 패턴 원리는 Spring Outbox 패턴 이벤트 발행 글을 참고하세요.

위로 스크롤
WordPress Appliance - Powered by TurnKey Linux