donghyeon.kang/clean-architecture-backend-template

Fork 0

Files

T

DongHyeonka e98b56eb03 feat: jpa, messaging, notification, mongo, graphql 어댑터터 리펙토링

2026-08-18 10:59:56 +09:00

73 KiB

Raw Permalink Blame History

Messaging Platform 설계서

상태: 구현 승인용 기준 설계
작성일: 2026-08-10
대상: Java/Spring Backend Skeleton의 공통 Messaging Capability
근거 자료: Java/Spring Messaging 플랫폼 심층 리서치
문서 역할: 구현 중 추가 설계 판단이 발생하지 않도록 공개 계약, 브로커별 보장, 오류·재시도·DLQ·Outbox·Inbox·운영·검증 경계를 고정한다.

1. 요약

이 플랫폼은 send()와 consume()을 편하게 감싸는 라이브러리가 아니다. 메시지 생성부터 브로커 수락, 복제·영속화, 소비자 전달, 처리, settlement, 재전달, DLQ, replay·redrive까지의 신뢰성 증거와 실패 의미론을 통제하는 백엔드 기반 모듈이다.

핵심 설계는 다음과 같다.

영역	확정 결정
공통 구조	자체 `messaging-core`가 공개 계약을 소유하고 브로커별 Native Adapter가 구현한다.
기본 API	M1 Typed Publisher·Handler를 일반 서비스의 기본 진입점으로 사용한다.
고급 API	M2 Batch·Manual Settlement·Pause/Resume·Delayed·Replay를 별도 권한으로 제공한다.
Native 기능	M3에서 Kafka transaction·partition, Rabbit routing·quorum, Pulsar subscription, NATS subject를 제한적으로 제공한다.
운영 기능	M4 Admin Plane에서 topology 검증, replay, redrive, offset reset, purge·delete를 제공한다.
전달 보장	`AT_MOST_ONCE`, `AT_LEAST_ONCE`만 공통 보장으로 제공한다.
Exactly-once	공통 `EXACTLY_ONCE` 옵션을 두지 않는다. 브로커 transaction 범위와 Inbox·Outbox 조합을 각각 명시한다.
publish 결과	성공·실패를 boolean으로 축소하지 않고 `CONFIRMED`, `REJECTED`, `AMBIGUOUS`와 broker evidence를 함께 반환한다.
소비자 기본	중복 전달을 정상적인 failure mode로 간주하고 handler 성공 뒤에만 settlement한다.
DLQ	DLQ publish가 confirm된 뒤 source를 settlement한다. DLQ publish 실패 시 source를 ACK하지 않는다.
Kafka	Kafka 4.2+를 Stable 기준으로 하고 4.3.x에서 검증한다. Share Group은 플랫폼 Experimental이다.
RabbitMQ	RabbitMQ 4.3.x를 Stable 기준으로 하고 durable work queue는 quorum queue를 기본값으로 한다.
Pulsar	4.0 LTS와 4.2 호환성을 Experimental Adapter에서 검증한다.
NATS	JetStream 2.14.x를 Experimental Adapter에서 검증한다. Core NATS는 Stable reliability 경로에 사용하지 않는다.
Reliability	Transactional Outbox, Inbox, Idempotent Consumer, Claim Check를 별도 Reliability 모듈로 제공한다.
Schema	JSON Stable, Avro·Protobuf 선택 Stable, Raw Bytes M2, Java Serialization 비지원이다.
CloudEvents	Domain·Integration Event에 선택 가능한 1.0.2 compatible profile을 제공한다.
topology	개발·테스트는 선택적 자동 생성, 운영은 IaC 생성 + 애플리케이션 startup 검증이 기본이다.
payload	portability 기본 상한은 1 MiB이며 초과 payload는 Claim Check를 사용한다.
관측성	논리 메시지와 물리 delivery·attempt를 분리해 측정하고 고카디널리티 값과 payload를 기록하지 않는다.

플랫폼의 완료 기준은 모든 브로커를 같아 보이게 만드는 것이 아니다. 다음 질문에 adapter가 동일한 형식으로 답할 수 있어야 한다.

publish가 실패했다.
→ broker에 전송되지 않았음을 아는가?
→ broker가 저장했을 가능성이 있는가?
→ 어떤 수준의 confirm을 받았는가?
→ 같은 logical message ID로 다시 시도해도 되는가?

consume이 실패했다.
→ handler가 시작됐는가?
→ side effect가 commit됐는가?
→ settlement가 broker에 반영됐는가?
→ redelivery가 발생할 수 있는가?

retry·DLQ·redrive를 수행한다.
→ 원래 message identity가 보존되는가?
→ ordering이 깨지는가?
→ duplicate side effect를 Inbox가 차단하는가?

2. 목표와 성공 기준

2.1 목표

일반 서비스가 브로커 종류를 몰라도 typed message를 publish·consume할 수 있게 한다.
브로커별 보장 차이를 숨기지 않고 capability와 evidence로 노출한다.
publish 결과가 확정적이지 않으면 AMBIGUOUS로 표현한다.
consumer duplicate, ACK 유실, rebalance, failover를 정상적인 운영 시나리오로 다룬다.
retry·DLQ·redrive에서 logical messageId와 schema identity를 보존한다.
Kafka와 RabbitMQ를 Stable Adapter로 제공한다.
Pulsar와 NATS JetStream을 Experimental Adapter로 제공해 abstraction boundary를 검증한다.
Outbox·Inbox를 실행 가능한 Reliability Recipe로 제공한다.
topology, 보안, 관측성, 장애·성능 검증을 코드와 CI에서 강제한다.

2.2 성공 기준

Core API에 broker SDK 타입과 Spring Message<?>가 노출되지 않는다.
Core API에 EXACTLY_ONCE enum이나 boolean 옵션이 존재하지 않는다.
모든 publish는 PublishResult 또는 안정 예외를 반환하며 AMBIGUOUS를 구분한다.
Kafka publish는 idempotent producer와 acks=all을 Stable profile에서 강제한다.
Rabbit publish는 publisher confirm과 routing outcome을 분리한다.
M1 consumer는 handler 성공 이전에 source settlement를 수행하지 않는다.
retry destination 또는 DLQ publish가 실패하면 source message를 settlement하지 않는다.
strict ordering destination에는 reorder 가능한 retry 전략을 설정할 수 없다.
Outbox relay가 ambiguous publish를 다시 시도해도 같은 messageId를 사용한다.
Inbox record와 business side effect가 같은 DB transaction에서 commit된다.
일반 애플리케이션 credential로 replay, redrive, purge, delete, offset reset을 호출할 수 없다.
동일 Core Contract Suite가 Kafka와 RabbitMQ에서 통과한다.
Experimental Adapter가 Core 계약을 변경하지 않고 Pulsar·NATS의 차이를 capability로 표현한다.
metric label과 log에 payload, secret header, 실제 message ID, 동적 key가 노출되지 않는다.

3. 입력 자료의 제약과 구현 가정

근거 자료에는 플랫폼 의미론과 지원 경계가 충분히 정의되어 있으나 실제 Backend Skeleton 저장소 구조와 dependency catalog는 포함되어 있지 않다. 구현 계획은 다음 가정을 사용한다.

항목	구현 가정
언어	Java 21
빌드	Gradle Kotlin DSL
root package	`io.backend.skeleton.messaging`
모듈 root	`modules/messaging`
Spring 버전	host 저장소의 dependency management를 사용하고 Spring Framework 6.2·7.0 호환 job을 둔다.
Core 비동기 타입	`CompletionStage`
Reactive facade	별도 Reactor adapter에서 `Mono`·`Flux`로 제공
DB recipe	PostgreSQL 16 + Spring JDBC/JPA + Flyway
JSON	Jackson
Avro·Protobuf	선택 모듈
관측성	Micrometer + OpenTelemetry exporter adapter
테스트	JUnit 5, AssertJ, ArchUnit, Testcontainers, Toxiproxy
코드 생성	사용하지 않는다. Schema code generation은 각 Avro·Protobuf 모듈에 한정한다.

실제 저장소 경로가 다르면 파일 경로만 기계적으로 매핑한다. 공개 타입, 모듈 의존 방향, 상태·오류·보장 계약은 변경하지 않는다.

다음 숫자는 플랫폼 starter의 초기 안전 기본값이다. 배포 profile에서 더 작게 조정할 수 있고, 더 크게 조정하려면 명시적 override와 성능 인증이 필요하다.

설정	기본값
logical payload 최대 크기	1,048,576 bytes
global hard payload 최대 크기	8,388,608 bytes
header 총 크기	32,768 bytes
header 개수	64
header key UTF-8 크기	128 bytes
header value UTF-8 크기	4,096 bytes
publish operation 기본 timeout	5초
consumer handler 기본 timeout	30초
graceful shutdown drain	30초
일반 destination 기본 retry 횟수	0회, 명시 profile만 허용
DLQ redrive batch	100건
Outbox relay batch	100건
Outbox lease	30초
Outbox polling	500ms

4. 범위

4.1 Core 포함 범위

Async Command
Domain Event
Integration Event
Work Queue
Publish–Subscribe
Event Stream
Typed Publish
Typed Handler
Batch Publish
Batch Consume
Manual Settlement
Pause / Resume
Delayed Delivery capability
Retry
DLQ / Parking
Replay / Redrive
Schema / Serialization
Broker Capability
Topology Validation
Security / Observability

4.2 별도 Reliability 모듈

Transactional Outbox
Inbox
Idempotent Consumer
Claim Check
Debezium Outbox Event Router integration

4.3 별도 기술 모듈

Email·SMS·Push               → notification
Browser live connection      → websocket
Redis Stream·Pub/Sub command → redis
Binary payload storage       → objectstorage / fileserver
CDC engine 자체              → connector/CDC
Saga orchestration           → 별도 orchestration

4.4 명시적 비지원

공통 EXACTLY_ONCE 설정
전역 순서 보장
DB와 broker의 자동 원자 transaction 보장
기본 XA
Java native serialization
무제한 payload·header
무한 retry
운영 application의 topology 파괴 작업
일반 애플리케이션에 raw broker client 반환
DLQ publish 확인 전 source ACK
one-shot payload의 자동 publish retry
replay·redrive를 일반 consumer API로 공개

5. 핵심 설계 원칙

증거를 보장 이름보다 우선한다. BROKER_ACK, REPLICATION_OR_PERSISTENCE_ACK, routing result를 별도 필드로 표현한다.
모호성을 숨기지 않는다. 전송 후 confirm을 받지 못하면 성공 또는 실패로 추정하지 않고 AMBIGUOUS를 반환한다.
중복 전달을 기본 전제로 둔다. consumer는 handler·DB·settlement 경계에서 duplicate-safe해야 한다.
logical identity를 유지한다. publish retry, redelivery, DLQ, redrive에서 messageId를 바꾸지 않는다.
ordering scope를 제한한다. 순서는 destination 전체가 아니라 partition·key·single consumer 같은 실제 범위로만 표현한다.
공통 API가 최소공배수가 되지 않게 한다. 공통 계약과 broker-native capability를 분리한다.
정책 우회 경로를 만들지 않는다. M2·M3도 TLS, credential, destination ACL, payload limit, trace, masking을 우회할 수 없다.
운영 변경은 Admin Plane에서 수행한다. 애플리케이션 runtime은 describe·validate만 수행한다.
자동 retry는 opt-in이다. 실패 분류, ordering, idempotency, deadline, broker capability가 모두 허용할 때만 수행한다.
DLQ도 publish다. DLQ publish 실패를 source success로 처리하지 않는다.
Outbox는 duplicate를 제거하지 않는다. stable message ID와 Inbox가 함께 있어야 effectively-once DB effect를 구성한다.
Stable과 Experimental을 코드로 분리한다. Experimental adapter가 Stable module dependency graph에 자동 유입되지 않는다.

6. 지원 매트릭스

6.1 브로커

브로커	등급	인증 기준	Stable 기능	제한·Experimental
Kafka	Stable	4.2+ / 4.3.x	producer, idempotence, consumer group, batch, pause/resume, replay, transaction capability	Share Group
RabbitMQ	Stable	4.3.x	exchange, routing, confirm, mandatory return, manual ACK, quorum queue, retry, DLQ	stream·특수 plugin
Pulsar	Experimental	4.0 LTS + 4.2	typed publish/consume, Shared, Key_Shared, schema	transaction·full compatibility 승격 전
NATS JetStream	Experimental	2.14.x	stream, durable consumer, explicit ACK, dedupe, replay	platform-managed DLQ workflow
Artemis/JMS	Extension	현재 Stable 범위 제외	adapter SPI만	별도 ADR과 Contract Suite 통과 뒤 지원 선언
Cloud broker	별도 Adapter	provider별	Core 변경 없이 adapter capability	IAM·visibility·ordering provider semantics

6.2 기능 등급

기능	등급
Typed Publish·Consume	Stable M1
At-least-once contract	Stable
Async publish result	Stable
Ambiguous publish	Stable
Auto settlement after success	Stable M1
Batch	M2
Manual settlement	M2
Request–Reply	M2, 제한적
Delayed·Scheduled	Capability-specific M2
Replay	M2 + Admin
Redrive	Admin
Broker transaction	M3 native capability
Kafka Share Group	Experimental M2/M3
Pulsar·NATS adapters	Experimental
Spring Cloud Stream bridge	Optional integration

7. 전체 아키텍처

Application
    │
    ├─ M1 Typed Messaging API
    │    ├─ MessagePublisher
    │    └─ MessageHandler
    │
    ├─ M2 Advanced API
    │    ├─ BatchPublisher
    │    ├─ ManualSettlementHandler
    │    ├─ PauseResumeController
    │    └─ ReplayRequest
    │
    ├─ M3 Native Capability
    │    ├─ KafkaNativeCapability
    │    ├─ RabbitNativeCapability
    │    ├─ PulsarNativeCapability
    │    └─ NatsNativeCapability
    │
    └─ M4 Admin Plane
         ├─ TopologyValidator
         ├─ ReplayService
         ├─ RedriveService
         └─ DestructiveOperationGuard

Policy Pipeline
    ├─ Envelope Validation
    ├─ Destination Resolution
    ├─ Schema Encoding
    ├─ Security / ACL
    ├─ Payload / Header Limit
    ├─ Retry / DLQ Policy
    ├─ Observability
    └─ Transport Adapter

Transport Adapters
    ├─ Kafka Stable
    ├─ RabbitMQ Stable
    ├─ Pulsar Experimental
    └─ NATS JetStream Experimental

Reliability
    ├─ Outbox
    ├─ Inbox
    ├─ Idempotent Consumer
    └─ Claim Check

7.1 Publish 흐름

Typed payload
→ MessageEnvelope 생성·검증
→ logical destination profile 조회
→ schema encoding
→ reserved header mapping
→ payload/header limit
→ broker adapter publish
→ broker evidence 수집
→ PublishResult 변환
→ metric·trace·audit 종료

7.2 Consume 흐름

broker delivery
→ transport metadata 추출
→ envelope/header 검증
→ schema decode
→ MessageDelivery 생성
→ handler 실행
→ success: source settlement
→ retry: policy에 따른 inline/pause/retry destination
→ permanent: confirmed DLQ publish 후 source settlement
→ settlement 결과·redelivery·lag 관측

8. 모듈 구조

modules/messaging/
├── messaging-core-api
├── messaging-schema-api
├── messaging-schema-json
├── messaging-schema-avro
├── messaging-schema-protobuf
├── messaging-cloudevents
├── messaging-policy
├── messaging-transport-spi
├── messaging-observability
├── messaging-security
├── messaging-kafka
├── messaging-kafka-share-experimental
├── messaging-rabbit
├── messaging-reliability-api
├── messaging-outbox-jpa
├── messaging-inbox-jpa
├── messaging-claim-check
├── messaging-admin-api
├── messaging-admin-runtime
├── messaging-pulsar-experimental
├── messaging-nats-experimental
├── messaging-spring-cloud-stream-bridge
├── messaging-spring-boot-starter
└── messaging-testkit

8.1 의존 방향

core-api
  ↑
policy / schema-api / transport-spi / observability / security
  ↑
kafka / rabbit / pulsar / nats
  ↑
spring-boot-starter

reliability-api
  ↑
outbox-jpa / inbox-jpa / claim-check

admin-api
  ↑
admin-runtime + broker adapter admin capability

금지 의존은 다음과 같다.

messaging-core-api → Spring Kafka, Spring AMQP, Pulsar, NATS
messaging-core-api → Spring Message<?>
Stable module → Experimental module
application API → M4 destructive implementation
Reliability core → 특정 broker SDK

9. 공개 API 등급

등급	사용 대상	제공 기능	금지 사항
M1	일반 업무 모듈	typed publish, typed handler, 자동 settlement	topic·offset·channel·delivery tag 직접 조작
M2	승인된 고급 모듈	batch, manual settlement, delayed, pause/resume, replay request	credential·payload limit·ACL 우회
M3	broker 특화 모듈	partition, routing, native transaction, subscription	raw client 반환, topology 파괴
M4	운영자·관리 서비스	topology, offset/cursor, redrive, purge/delete	application credential 사용

M2와 M3는 @RequiresMessagingCapability 또는 명시적 bean qualifier를 통해서만 주입된다. M4는 별도 Spring Boot application 또는 별도 security chain과 credential을 사용한다.

10. 핵심 식별자와 타입

public record MessageId(UUID value) {
    public static MessageId newId() {
        return new MessageId(UuidV7.next());
    }
}

public record MessageType(String value) {}
public record SchemaVersion(int value) {}
public record ProducerId(String value) {}
public record DestinationName(String value) {}
public record CorrelationId(String value) {}
public record CausationId(MessageId value) {}
public record RedriveId(UUID value) {}
public record ReplayId(UUID value) {}

10.1 식별자 규칙

messageId는 logical message identity다.
publish retry, broker redelivery, retry destination, DLQ, redrive에서 유지한다.
새 업무 사건 또는 새 command일 때만 새 messageId를 생성한다.
redriveId, replayId, transportAttemptId는 별도 식별자다.
Java class name을 messageType으로 사용하지 않는다.
messageType은 order.created, payment.capture.requested와 같은 안정된 catalog 값이다.
실제 messageId는 metric label에 사용하지 않는다.

11. Message Envelope

public record MessageEnvelope<T>(
        MessageId messageId,
        MessageType messageType,
        SchemaVersion schemaVersion,
        Instant producedAt,
        Optional<Instant> occurredAt,
        ProducerId producer,
        Optional<CorrelationId> correlationId,
        Optional<CausationId> causationId,
        ContentType contentType,
        Optional<String> partitionKey,
        Optional<String> orderingKey,
        Optional<TenantContext> tenantContext,
        TraceContext traceContext,
        MessageHeaders headers,
        T payload) {
}

11.1 필드 정책

필드	정책
`messageId`	필수, UUIDv7
`messageType`	필수, catalog 등록
`schemaVersion`	1 이상
`producedAt`	platform이 설정
`occurredAt`	Event 필수, Command 선택
`producer`	logical service ID
`correlationId`	workflow 단위
`causationId`	선행 message가 있을 때
`contentType`	codec registry 값
`partitionKey`	distribution 목적
`orderingKey`	ordering 의도
`tenantContext`	allowlist된 bounded value
`traceContext`	platform이 생성·전파
`headers`	제한형 immutable map
`payload`	null 금지. Kafka tombstone은 M3 전용 타입

11.2 Header 정책

public final class MessageHeaders {
    private final Map<HeaderName, HeaderValue> values;

    public static MessageHeaders of(Map<HeaderName, HeaderValue> values) {
        return HeaderPolicy.validateAndCopy(values);
    }
}

예약 header는 platform만 쓸 수 있다.

msg.id
msg.type
msg.schema-version
msg.producer
msg.produced-at
msg.occurred-at
msg.correlation-id
msg.causation-id
msg.content-type
msg.partition-key
msg.ordering-key
msg.redrive-id
msg.redrive-count
traceparent
tracestate
baggage

금지 header는 다음과 같다.

Authorization
Proxy-Authorization
Cookie
Set-Cookie
access_token
refresh_token
api_key
password
client_secret

12. 논리 Destination과 Profile

public record MessageDestination<T>(
        DestinationName name,
        MessageType messageType,
        Class<T> payloadType) {
}

애플리케이션은 topic, exchange, queue, subject가 아니라 logical destination을 사용한다. 물리 매핑은 DestinationProfile이 소유한다.

messaging:
  destinations:
    order-events:
      broker: kafka-primary
      kind: EVENT_STREAM
      physical:
        topic: order.events.v1
      schema:
        codec: json
        compatibility: BACKWARD_TRANSITIVE
      guarantees:
        delivery: AT_LEAST_ONCE
        ordering: KEY
        external-side-effect: INBOX_TRANSACTIONAL
      producer:
        confirmation: REPLICATION_OR_PERSISTENCE_ACK
        timeout: 5s
      consumer:
        group: order-projection
        concurrency: 6
        max-in-flight-per-partition: 1
        handler-timeout: 30s
      retry:
        mode: PAUSE_PARTITION
        max-attempts: 3
        initial-delay: 200ms
        max-delay: 2s
        ordering-impact: PRESERVE
      dlq:
        destination: order-events-dlq
      payload:
        max-bytes: 1048576

    email-work:
      broker: rabbit-primary
      kind: WORK_QUEUE
      physical:
        exchange: notification.work
        routing-key: email
        queue: notification.email.q
      guarantees:
        delivery: AT_LEAST_ONCE
        ordering: NONE
        external-side-effect: IDEMPOTENCY_REQUIRED
      producer:
        confirmation: REPLICATION_OR_PERSISTENCE_ACK
        mandatory-routing: true
        timeout: 5s
      consumer:
        concurrency: 8
        prefetch: 16
        handler-timeout: 30s
      retry:
        mode: RETRY_DESTINATION
        max-attempts: 5
        initial-delay: 1s
        max-delay: 5m
        ordering-impact: ALLOW_REORDER
      dlq:
        destination: email-work-dlq
      payload:
        max-bytes: 1048576

12.1 Startup validation

다음 조건이면 애플리케이션 시작을 실패시킨다.

AT_LEAST_ONCE인데 adapter가 confirm 또는 explicit settlement를 제공하지 못함
ordering-impact=PRESERVE인데 retry destination 전략 사용
ordering=KEY인데 key resolver 없음
M1 destination에 manual settlement 설정
payload hard limit 초과
DLQ가 자기 자신을 가리킴
retry destination cycle 존재
운영 profile에서 topology auto-create 활성화
production에서 TLS 또는 broker 인증 비활성
Kafka Stable producer에서 idempotence 또는 acks=all 비활성
Rabbit durable work queue에서 classic transient queue 사용
schema codec 또는 message type catalog 누락

13. Capability Model

공통 API는 브로커가 모든 기능을 동일하게 지원한다고 가정하지 않는다.

public record MessagingCapabilities(
        boolean brokerAcknowledgement,
        boolean replicationOrPersistenceEvidence,
        boolean perMessageSettlement,
        boolean batchSettlement,
        boolean orderedStream,
        boolean keyedOrdering,
        boolean replay,
        boolean delayedDelivery,
        boolean brokerTransaction,
        boolean deduplicatedPublish,
        boolean nativeDeadLetter,
        boolean topologyManagement) {
}

실행 시에는 destination별로 capability snapshot을 만든다.

public interface CapabilityRegistry {
    DestinationCapabilities capabilities(DestinationName destination);
}

13.1 Capability 검증 규칙

요청 기능	필수 capability
`confirmation=REPLICATION_OR_PERSISTENCE_ACK`	adapter가 해당 증거를 입증
manual settlement	`perMessageSettlement` 또는 명시적 batch settlement
replay	`replay`
broker transaction	`brokerTransaction`
strict key ordering	`keyedOrdering` + concurrency 제약
delayed delivery	native 또는 retry destination 구현
native DLQ	native dead-letter 보장 조건 충족
deduplicated publish	broker dedupe + stable message ID

Capability가 없으면 조용히 기능을 약화하지 않고 startup error 또는 MessagingCapabilityUnavailableException을 반환한다.

14. Schema·직렬화 계약

14.1 Codec API

public interface MessageCodec {
    ContentType contentType();

    EncodedMessage encode(
            MessageType type,
            SchemaVersion version,
            Object payload);

    <T> T decode(
            MessageType type,
            SchemaVersion version,
            EncodedMessage encoded,
            Class<T> payloadType);
}

public record EncodedMessage(
        byte[] bytes,
        ContentType contentType,
        Optional<SchemaReference> schemaReference) {
}

14.2 지원 수준

형식	등급	정책
JSON	Stable 기본	Jackson allowlist, 깊이·크기 제한
Avro	선택 Stable	Registry profile과 compatibility gate
Protobuf	선택 Stable	generated type + unknown field 정책
JSON Schema	Registry profile	JSON codec와 연동
Raw Bytes	M2	schema 검증 우회 감사
Java Serialization	비지원	wire compatibility·보안 문제

14.3 Compatibility

BACKWARD
BACKWARD_TRANSITIVE
FORWARD
FORWARD_TRANSITIVE
FULL
FULL_TRANSITIVE
NONE_EXPERIMENTAL

운영 기본값은 다음과 같다.

integration event: BACKWARD_TRANSITIVE
장수 공용 event: FULL_TRANSITIVE 검토
command: producer·consumer 동시 배포가 가능하면 BACKWARD
schema 없음: M2 Raw Bytes에서만 허용

14.4 Schema 변경 규칙

optional/default 없는 필드 추가를 금지한다.
필드 rename은 add → dual read/write → remove로 처리한다.
enum에는 unknown value 전략이 있어야 한다.
message type 변경은 새 type으로 취급한다.
역직렬화 실패는 일반 transient retry 대상이 아니다.
Kafka tombstone은 nullable payload로 일반화하지 않고 KafkaTombstonePublish M3 API로 분리한다.

14.5 Golden Message

각 message type은 다음 fixture를 보유한다.

schemas/<message-type>/v1/valid.*
schemas/<message-type>/v2/valid.*
schemas/<message-type>/invalid/*

CI에서 구버전 fixture를 최신 consumer가 읽고, compatibility mode에 따라 신버전 fixture를 구버전 consumer가 읽는지 검증한다.

15. CloudEvents Profile

CloudEvents는 Domain·Integration Event의 상호운용 profile로만 제공한다. Command와 Work Item에는 강제하지 않는다.

public interface CloudEventMapper {
    CloudEvent toCloudEvent(MessageEnvelope<?> envelope, URI source);
    MessageEnvelope<EncodedMessage> fromCloudEvent(CloudEvent event);
}

매핑은 다음과 같다.

Envelope	CloudEvents
`messageId`	`id`
`producer/source`	`source`
`messageType`	`type`
`occurredAt`	`time`
`contentType`	`datacontenttype`
schema URI	`dataschema`
`correlationId`	extension `correlationid`
`causationId`	extension `causationid`
`schemaVersion`	extension `schemaversion`
`tenantContext`	extension `tenantcontext`

Kafka binary binding을 사용할 때 tombstone과 empty event를 구분한다. CloudEvent data가 없다는 이유로 Kafka null value를 자동 생성하지 않는다.

16. Publisher 공개 계약

public interface MessagePublisher {
    <T> CompletionStage<PublishResult> publish(
            MessageDestination<T> destination,
            MessageEnvelope<T> message,
            PublishOptions options);
}

public record PublishOptions(
        Duration timeout,
        ConfirmationRequirement confirmation,
        Optional<PublishDeduplication> deduplication,
        Map<String, String> brokerHints) {
}

M1에서는 brokerHints가 비어 있어야 한다. M3 adapter만 typed hint를 추가할 수 있다.

16.1 Blocking·Reactive facade

public interface BlockingMessagePublisher {
    <T> PublishResult publish(
            MessageDestination<T> destination,
            MessageEnvelope<T> message,
            PublishOptions options);
}

public interface ReactiveMessagePublisher {
    <T> Mono<PublishResult> publish(
            MessageDestination<T> destination,
            MessageEnvelope<T> message,
            PublishOptions options);
}

Core 구현은 CompletionStage를 사용하고 Blocking·Reactive facade가 lifecycle과 cancellation을 변환한다.

16.2 Batch Publish

public interface BatchMessagePublisher {
    CompletionStage<BatchPublishResult> publish(
            List<PublishRequest<?>> requests,
            BatchPublishOptions options);
}

Batch는 transaction이 아니다. 결과는 입력 index별 성공·실패·모호성을 유지한다.

public record BatchPublishResult(
        List<BatchPublishItemResult> items,
        Duration elapsed) {
}

자동 batch retry는 하지 않는다. 호출자가 duplicate-safe한 item만 같은 messageId로 다시 제출한다.

17. Publish 결과와 증거

17.1 결과 모델

public record PublishResult(
        PublishCompletion completion,
        PublishEvidence evidence,
        RoutingOutcome routingOutcome,
        Optional<BrokerPosition> position,
        int attempts,
        Duration elapsed,
        Optional<FailureDescriptor> failure) {
}

public enum PublishCompletion {
    CONFIRMED,
    REJECTED,
    AMBIGUOUS
}

public record PublishEvidence(
        boolean queuedLocally,
        TransmissionEvidence transmission,
        boolean brokerAccepted,
        ConfirmationLevel confirmationLevel) {
}

public enum TransmissionEvidence {
    NOT_TRANSMITTED,
    MAY_HAVE_BEEN_TRANSMITTED,
    TRANSMITTED
}

public enum ConfirmationLevel {
    NONE,
    BROKER_ACK,
    REPLICATION_OR_PERSISTENCE_ACK
}

public enum RoutingOutcome {
    NOT_APPLICABLE,
    ROUTED,
    UNROUTABLE,
    UNKNOWN
}

17.2 판정 규칙

상황	결과
로컬 validation 실패	`REJECTED`, `NOT_TRANSMITTED`
broker 명시적 reject	`REJECTED`
confirm 수신	`CONFIRMED`
Rabbit confirm + unroutable return	`REJECTED`, `UNROUTABLE`
bytes 전송 후 connection loss	`AMBIGUOUS`
confirm timeout	`AMBIGUOUS`
broker ACK 수준이 요구보다 약함	`REJECTED` 또는 startup validation failure
adapter가 증거를 판정할 수 없음	보수적으로 `AMBIGUOUS`

17.3 Broker Position

public interface BrokerPosition {
    String broker();
    Map<String, String> diagnosticAttributes();
}

public record KafkaPosition(
        String topic,
        int partition,
        long offset) implements BrokerPosition {
    @Override public String broker() { return "kafka"; }
    @Override public Map<String, String> diagnosticAttributes() {
        return Map.of("topic", topic, "partition", Integer.toString(partition),
                "offset", Long.toString(offset));
    }
}

public record RabbitPublishReference(
        String exchange,
        String routingKey,
        long sequence) implements BrokerPosition {
    @Override public String broker() { return "rabbitmq"; }
    @Override public Map<String, String> diagnosticAttributes() {
        return Map.of("exchange", exchange, "routingKey", routingKey,
                "sequence", Long.toString(sequence));
    }
}

Position은 진단·replay에 쓰며 공통 업무 로직의 분기 기준으로 사용하지 않는다.

18. Producer 상태 머신

CREATED
  ↓
VALIDATED
  ↓
ENCODED
  ↓
QUEUED_LOCALLY
  ↓
TRANSMITTING
  ├─→ REJECTED
  ├─→ AMBIGUOUS
  ↓
BROKER_ACCEPTED
  ├─→ CONFIRMED
  └─→ AMBIGUOUS

상태 머신은 내부 trace·failure evidence에 사용한다. 공개 결과에는 최종 completion과 evidence만 노출한다.

18.1 자동 publish retry

Core publisher의 기본은 자동 retry 없음이다. adapter 내부 protocol retry는 다음 조건에서만 허용한다.

동일 messageId 유지
broker-native dedup 또는 producer idempotence 활성
payload replayable
confirmation timeout 전체 예산 내
adapter가 duplicate risk를 문서화

AMBIGUOUS 결과를 새 messageId로 자동 재발행하지 않는다.

19. Consumer 공개 계약

public interface MessageHandler<T> {
    CompletionStage<HandleResult> handle(MessageDelivery<T> delivery);
}

public record MessageDelivery<T>(
        MessageEnvelope<T> message,
        DeliveryMetadata metadata,
        DeliveryContext context) {
}

public sealed interface HandleResult
        permits HandleResult.Success,
                HandleResult.Retry,
                HandleResult.DeadLetter,
                HandleResult.Reject {

    record Success() implements HandleResult {}

    record Retry(FailureDescriptor failure) implements HandleResult {}

    record DeadLetter(FailureDescriptor failure) implements HandleResult {}

    record Reject(FailureDescriptor failure) implements HandleResult {}
}

19.1 Delivery Metadata

public record DeliveryMetadata(
        DestinationName destination,
        int deliveryAttempt,
        boolean redelivered,
        Optional<BrokerPosition> brokerPosition,
        Optional<String> partitionOrQueue,
        Optional<String> consumerGroup,
        Instant receivedAt) {
}

19.2 M1 규칙

handler는 broker ACK API를 호출하지 않는다.
Success 후에만 source settlement한다.
Retry는 destination policy가 전략을 결정한다.
DeadLetter는 DLQ publish confirm 후 source settlement한다.
Reject는 명시적으로 discard를 허용한 at-most-once profile에서만 허용한다.
handler timeout은 PROCESSING_TRANSIENT로 분류하되 side effect commit 여부가 불명확하면 duplicate 가능성을 기록한다.

20. M2 Manual Settlement

public interface ManualMessageHandler<T> {
    CompletionStage<Void> handle(
            MessageDelivery<T> delivery,
            SettlementController settlement);
}

public interface SettlementController {
    CompletionStage<SettlementResult> ack();
    CompletionStage<SettlementResult> retry(Duration delay);
    CompletionStage<SettlementResult> deadLetter(FailureDescriptor failure);
    CompletionStage<SettlementResult> reject(FailureDescriptor failure);
}

20.1 Batch Consume

public interface BatchMessageHandler<T> {
    CompletionStage<HandleResult> handle(BatchMessageDelivery<T> batch);
}

public record BatchMessageDelivery<T>(
        List<MessageDelivery<T>> deliveries,
        BatchDeliveryMetadata metadata) {
}

Batch consume은 M2다. batch 전체 settlement가 가능한 broker에서는 전체 결과를 사용하고, 개별 settlement가 가능한 broker에서는 item 결과를 유지한다. strict ordering destination에서는 하나의 ordering unit을 넘는 batch를 허용하지 않는다.

20.2 Pause·Resume와 Delayed Publish

public interface PauseResumeController {
    CompletionStage<Void> pause(DestinationName destination, String scope);
    CompletionStage<Void> resume(DestinationName destination, String scope);
}

public interface DelayedMessagePublisher {
    <T> CompletionStage<PublishResult> publish(
            MessageDestination<T> destination,
            MessageEnvelope<T> message,
            Instant deliverAt);
}

두 API 모두 M2이며 destination capability가 없으면 명시적으로 거부한다. Kafka 일반 topic은 native scheduled delivery capability를 선언하지 않는다. Rabbit delayed/quorum retry, Pulsar delayed delivery, NATS scheduler 계열은 adapter capability로 제공한다.

Manual API는 다음 guard를 강제한다.

정확히 한 번만 terminal settlement 호출
handler 종료 전 settlement 누락 감지
source ACK 전 DLQ confirmation
application thread에서 broker channel·consumer object 접근 금지
settlement timeout과 SETTLEMENT_UNKNOWN 구분
duplicate call은 SettlementAlreadyCompletedException

21. Consumer 상태와 Settlement 증거

RECEIVED
  ↓
DECODING
  ├─→ SCHEMA_PARKING
  ↓
PROCESSING
  ├─→ RETRY_PENDING
  ├─→ DEAD_LETTER_PENDING
  ├─→ REJECTED
  ↓
HANDLER_SUCCEEDED
  ↓
SETTLEMENT_SENDING
  ├─→ SETTLED
  └─→ SETTLEMENT_UNKNOWN

Reliability module을 사용할 때만 다음 증거를 추가한다.

INBOX_RESERVED
BUSINESS_TRANSACTION_COMMITTED

21.1 Settlement 결과

public record SettlementResult(
        SettlementCompletion completion,
        SettlementEvidence evidence,
        Optional<FailureDescriptor> failure) {
}

public enum SettlementCompletion {
    SETTLED,
    REJECTED,
    UNKNOWN
}

UNKNOWN은 redelivery 가능성을 의미한다. 호출자는 이를 성공으로 간주하지 않는다.

22. 전달·처리·순서 보장 모델

public enum DeliveryGuarantee {
    AT_MOST_ONCE,
    AT_LEAST_ONCE
}

public enum ProcessingGuarantee {
    APPLICATION_IDEMPOTENT,
    BROKER_TRANSACTIONAL
}

public enum OrderingScope {
    NONE,
    DESTINATION,
    PARTITION,
    KEY
}

public enum ExternalSideEffectGuarantee {
    NONE,
    IDEMPOTENCY_REQUIRED,
    INBOX_TRANSACTIONAL
}

22.1 금지 모델

다음은 Core에 존재하지 않는다.

boolean exactlyOnce;
DeliveryGuarantee.EXACTLY_ONCE;
OrderingScope.GLOBAL;

22.2 실제 의미

모델	의미
`AT_MOST_ONCE`	유실 가능, 중복 억제 우선
`AT_LEAST_ONCE`	redelivery 가능, 유실 방지 우선
`APPLICATION_IDEMPOTENT`	handler가 message ID 또는 업무 key로 중복 effect를 막음
`BROKER_TRANSACTIONAL`	broker가 정의한 transaction 범위에서만 원자화
`INBOX_TRANSACTIONAL`	Inbox row와 DB side effect가 같은 transaction
`PARTITION`	한 partition 안의 broker order
`KEY`	동일 key의 mapping이 유지되는 동안의 순서

22.3 Ordering validator

다음 조합을 거부한다.

ordering=KEY + key resolver 없음
ordering=PARTITION + retry topic으로 reorder 허용 안 함
strict order + concurrency가 실제 ordering unit보다 큼
NATS ordered consumer + competing work queue 설정
Pulsar Shared + key ordering 요구
Kafka Share Group + ordered stream 요구

23. 오류 모델

MessagingException
 ├─ MessagingConfigurationException
 ├─ MessagingCapabilityUnavailableException
 ├─ MessageValidationException
 ├─ MessageSerializationException
 ├─ MessageSchemaIncompatibleException
 ├─ MessageTooLargeException
 ├─ MessageHeaderRejectedException
 ├─ MessagePublishRejectedException
 ├─ MessagePublishAmbiguousException
 ├─ MessagePublishTimeoutException
 ├─ MessageRoutingException
 ├─ MessageAuthenticationException
 ├─ MessageAuthorizationException
 ├─ MessageConsumerException
 ├─ MessageHandlerTimeoutException
 ├─ MessageSettlementException
 ├─ MessageSettlementUnknownException
 ├─ MessageRetryExhaustedException
 ├─ MessageDeadLetterException
 ├─ MessageRedriveException
 ├─ MessageTopologyException
 └─ MessageBrokerUnavailableException

23.1 Failure Category

public enum FailureCategory {
    TRANSIENT_INFRASTRUCTURE,
    THROTTLED,
    PROCESSING_TRANSIENT,
    PERMANENT_BUSINESS,
    POISON_MESSAGE,
    DESERIALIZATION,
    AUTHENTICATION,
    AUTHORIZATION,
    AMBIGUOUS,
    CONFIGURATION
}

23.2 Failure Descriptor

public record FailureDescriptor(
        FailureCategory category,
        String code,
        boolean retryable,
        String sanitizedMessage,
        Optional<String> exceptionType) {
}

Payload, full stack trace, credential, actual message key는 message header에 적재하지 않는다. Stack trace는 secure log storage에만 기록한다.

24. Retry Policy Engine

24.1 정책 모델

public record RetryPolicy(
        RetryMode mode,
        int maxAttempts,
        Duration initialDelay,
        Duration maxDelay,
        double multiplier,
        boolean jitter,
        OrderingImpact orderingImpact,
        Set<FailureCategory> retryableCategories,
        Set<FailureCategory> nonRetryableCategories) {
}

public enum RetryMode {
    NONE,
    INLINE,
    BLOCKING,
    PAUSE_PARTITION,
    RETRY_DESTINATION,
    BROKER_DELAYED
}

public enum OrderingImpact {
    PRESERVE,
    ALLOW_REORDER
}

24.2 결정 입력

public record RetryContext(
        DestinationProfile destination,
        DeliveryMetadata delivery,
        FailureDescriptor failure,
        MessagingCapabilities capabilities,
        boolean handlerMayHaveCommittedSideEffect) {
}

24.3 결정 결과

public sealed interface RetryDecision {
    record RetryInline(Duration delay) implements RetryDecision {}
    record PauseAndRetry(Duration delay) implements RetryDecision {}
    record PublishToRetryDestination(
            DestinationName destination,
            Duration delay) implements RetryDecision {}
    record DeadLetter(FailureDescriptor failure) implements RetryDecision {}
    record Reject(FailureDescriptor failure) implements RetryDecision {}
}

24.4 기본 규칙

DESERIALIZATION, AUTHENTICATION, AUTHORIZATION, CONFIGURATION은 자동 retry하지 않는다.
PERMANENT_BUSINESS는 DLQ 또는 reject 정책으로 보낸다.
TRANSIENT_INFRASTRUCTURE, THROTTLED, PROCESSING_TRANSIENT만 기본 retry 후보이다.
attempt는 최초 delivery를 1로 계산한다.
max attempts를 넘으면 DLQ 또는 parking으로 전환한다.
retry destination으로 이동해도 messageId를 유지한다.
retryAttempt, firstFailureAt, lastFailureAt은 reserved metadata에 기록한다.
source settlement는 retry destination publish confirmation 이후에만 수행한다.
strict ordering에서는 PAUSE_PARTITION 또는 blocking 전략만 허용한다.
Retry 중에도 consumer shutdown이 시작되면 신규 attempt를 생성하지 않는다.

25. DLQ·Parking·Redrive

25.1 DLQ 처리 순서

handler permanent failure 또는 retry exhausted
→ DeadLetterEnvelope 생성
→ DLQ publish
→ DLQ broker confirmation 확인
→ source settlement

DLQ publish가 실패하면 source를 ACK하지 않고 consumer를 pause하거나 원래 retry 정책으로 되돌린다.

25.2 Dead Letter Envelope

public record DeadLetterMetadata(
        MessageId originalMessageId,
        DestinationName originalDestination,
        Optional<BrokerPosition> originalPosition,
        Instant firstFailureAt,
        Instant lastFailureAt,
        int attempts,
        FailureCategory failureCategory,
        String failureCode,
        int redriveCount,
        Optional<RedriveId> lastRedriveId) {
}

원본 payload와 schema identity를 유지한다. 전체 stack trace는 header에 넣지 않는다.

25.3 Parking

역직렬화 불가, schema 미지원, 관리자 판정이 필요한 메시지는 일반 DLQ와 분리된 parking destination에 저장한다.

schema-parking
security-parking
manual-review-parking

25.4 Redrive

public record RedriveRequest(
        DestinationName sourceDlq,
        DestinationName target,
        Set<MessageId> messageIds,
        String operator,
        String reason,
        boolean dryRun) {
}

Redrive 규칙은 다음과 같다.

기존 messageId 유지
새 redriveId 생성
redriveCount 증가
대상 schema와 destination capability 재검증
dry-run 기본
운영자·사유·선택 범위를 audit
동일 redrive request의 idempotency 보장
redrive 성공 confirmation 후 DLQ 원본을 mark 또는 settlement

26. Kafka Stable Adapter

26.1 Stable 기준

Kafka 4.2 이상
4.3.x release gate
Spring Kafka native integration
traditional consumer group Stable
Share Group은 별도 Experimental module

26.2 Producer 설정

Stable profile은 다음을 강제한다.

enable.idempotence=true
acks=all
retries > 0
max.in.flight.requests.per.connection <= 5

플랫폼 timeout과 Kafka delivery timeout의 관계를 startup에서 검증한다.

platform publish timeout
  >= request.timeout.ms
  <= delivery.timeout.ms + platform cleanup margin

ProducerRecord에는 logical messageId, schema metadata, trace context를 header로 기록한다.

26.3 Publish evidence

Kafka 결과	플랫폼 매핑
`RecordMetadata` 수신	`CONFIRMED`, `BROKER_ACK` 또는 profile 조건 충족 시 stronger evidence
serializer failure	`REJECTED`, `NOT_TRANSMITTED`
authorization failure	`REJECTED`
producer fenced	`REJECTED`, non-retry
delivery timeout	전송 여부에 따라 `AMBIGUOUS`
connection loss after send	`AMBIGUOUS`

Kafka acks=all을 영구 보존으로 표현하지 않는다. confirmation level은 현재 ISR acknowledgement 증거라는 adapter-specific detail을 diagnostic metadata에 남긴다.

26.4 Consumer Group

Stable consumer는 다음을 사용한다.

enable.auto.commit=false
isolation.level=read_committed  // transactional destination profile

M1의 strict order path는 partition별 한 개의 in-flight handler를 기본으로 한다.

poll
→ record를 partition work coordinator에 전달
→ partition pause
→ bounded worker에서 handler 실행
→ settlement command queue에 결과 기록
→ consumer poll thread가 command를 drain
→ contiguous offset commit
→ partition resume

이 구조는 handler thread가 Kafka consumer를 직접 호출하지 않게 한다.

26.5 Offset Tracker

public interface PartitionOffsetTracker {
    void delivered(TopicPartition partition, long offset);
    void completed(TopicPartition partition, long offset);
    OptionalLong highestContiguousCompleted(TopicPartition partition);
}

parallel handler를 허용해도 commit은 완료된 contiguous offset까지만 진행한다.

26.6 Retry

요구	전략
partition order 보존	pause partition + bounded retry
처리량 우선, reorder 허용	retry topic
짧은 transient failure	inline retry
poison	DLT

Retry topic 사용 시 원래 ordering을 보장하지 않는다고 profile과 metric에 표시한다.

26.7 Transaction Capability

Kafka transaction은 M3로 제공한다.

public interface KafkaTransactionalProcessor<K, V, R> {
    CompletionStage<R> process(
            KafkaTransactionalDelivery<K, V> delivery,
            KafkaTransactionalPublisher publisher);
}

보장 범위는 Kafka input offset과 Kafka output record다. 외부 DB·HTTP side effect는 포함하지 않는다.

26.8 Replay

Replay는 M2/Admin에서만 제공한다.

seek by offset
seek by timestamp
replay to isolated consumer group
replay to new destination

기존 production consumer group offset reset은 M4 위험 작업이다.

26.9 Tombstone

public interface KafkaTombstonePublisher {
    CompletionStage<PublishResult> publishTombstone(
            DestinationName destination,
            String key,
            MessageHeaders headers);
}

일반 MessageEnvelope의 null payload로 대체하지 않는다.

Kafka 4.2+ Share Group은 record 단위 acknowledgement와 delivery attempt counting을 제공하지만 ordered stream abstraction으로 사용하지 않는다.

public interface KafkaShareWorkQueueCapability {
    <T> void register(
            MessageDestination<T> destination,
            MessageHandler<T> handler,
            ShareGroupOptions options);
}

제약은 다음과 같다.

OrderingScope.NONE만 허용
Work Queue destination에서만 사용
platform settlement mapping이 안정화되기 전 Experimental
traditional consumer group과 동일한 transaction API를 가정하지 않음
redelivery attempt와 share acknowledgement를 별도 native contract test로 검증

28. RabbitMQ Stable Adapter

28.1 Stable 기준

RabbitMQ 4.3.x
Spring AMQP native integration
durable work queue는 quorum queue 기본
publisher confirm과 consumer ACK 분리

28.2 Producer

Stable producer는 다음을 강제한다.

publisher-confirm-type=correlated
publisher-returns=true
mandatory=true

confirm과 return을 correlation ID로 결합한다.

public record RabbitPublishOutcome(
        RabbitConfirmOutcome confirm,
        RoutingOutcome routing,
        Optional<String> replyCode,
        Optional<String> replyText) {
}

상태	플랫폼 결과
confirm ACK + routed	`CONFIRMED`
confirm ACK + returned	`REJECTED`, `UNROUTABLE`
confirm NACK	`REJECTED`
channel close before result	`AMBIGUOUS`
confirm timeout	`AMBIGUOUS`

28.3 Consumer

manual acknowledgement
prefetch로 in-flight 제한
handler success 후 ACK
transient failure는 requeue 또는 retry queue 정책
permanent failure는 confirmed DLQ publish 후 source ACK
channel·delivery tag를 application handler에 노출하지 않음

28.4 Queue 기본값

durable=true
auto-delete=false
exclusive=false
queue-type=quorum

classic queue는 명시적 low-durability profile에서만 허용한다.

28.5 Ordering

Rabbit queue는 enqueue FIFO를 출발점으로 하지만 다음 조건에서 처리 완료 순서는 달라질 수 있다.

consumer concurrency > 1
priority queue
NACK / requeue
redelivery
retry queue

따라서 Stable common contract는 OrderingScope.DESTINATION을 자동 선언하지 않는다. strict ordering profile은 single active consumer와 concurrency 1을 요구한다.

28.6 DLQ

두 모드를 제공한다.

Platform-managed DLQ
- target publish confirm 후 source ACK
- broker-neutral 기본
Native quorum at-least-once dead-lettering
- M3 capability
- 필요한 queue policy를 startup에서 검증

기본 dead-letter strategy가 at-most-once인 topology를 stronger guarantee로 표현하지 않는다.

28.7 Retry

inline retry
retry queue + TTL / dead-letter routing
RabbitMQ 4.3 quorum delayed retry capability

retry cycle은 x-death 또는 platform attempt metadata로 제한한다.

28.8 Request–Reply

M2에서만 제공한다.

correlation ID 필수
reply timeout 필수
temporary reply queue lifecycle 관리
장기 RPC 대체로 사용하지 않음
duplicate request·late reply 처리 정의

29. Reliability API

public interface ReliableMessagePublisher {
    <T> void addToOutbox(
            MessageDestination<T> destination,
            MessageEnvelope<T> message);
}

public interface IdempotentMessageHandler<T> {
    CompletionStage<HandleResult> handleOnce(
            String consumerName,
            MessageDelivery<T> delivery,
            TransactionalMessageAction<T> action);
}

Reliability module은 Core의 publish·delivery 계약을 바꾸지 않는다. DB transaction과 message identity를 결합한다.

30. Transactional Outbox

30.1 테이블

create table messaging_outbox (
    id uuid primary key,
    message_id uuid not null unique,
    destination varchar(160) not null,
    message_type varchar(240) not null,
    schema_version integer not null,
    content_type varchar(120) not null,
    aggregate_type varchar(160),
    aggregate_id varchar(320),
    headers jsonb not null,
    payload bytea not null,
    payload_hash varchar(128) not null,
    occurred_at timestamptz,
    created_at timestamptz not null,
    available_at timestamptz not null,
    status varchar(40) not null,
    attempts integer not null default 0,
    next_attempt_at timestamptz,
    lease_owner varchar(160),
    lease_until timestamptz,
    last_failure_category varchar(80),
    last_failure_code varchar(160),
    published_at timestamptz,
    version bigint not null default 0
);

create index ix_messaging_outbox_poll
    on messaging_outbox(status, available_at, next_attempt_at, created_at);

create index ix_messaging_outbox_lease
    on messaging_outbox(lease_until)
    where lease_owner is not null;

30.2 상태

PENDING
CLAIMED
PUBLISHED
RETRYABLE_FAILURE
PARKED

30.3 Relay 알고리즘

SELECT id FROM messaging_outbox
WHERE status IN ('PENDING', 'RETRYABLE_FAILURE')
  AND available_at <= now()
  AND coalesce(next_attempt_at, available_at) <= now()
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT :batchSize
→ lease와 attempt 증가
→ DB transaction commit
→ Core publisher로 같은 messageId publish
→ CONFIRMED: PUBLISHED
→ REJECTED retryable: RETRYABLE_FAILURE
→ permanent: PARKED
→ AMBIGUOUS: RETRYABLE_FAILURE, 같은 messageId로 재시도

Outbox가 duplicate publish를 막는다고 광고하지 않는다. ambiguous retry는 같은 messageId를 사용하고 downstream Inbox가 duplicate side effect를 차단한다. PUBLISHED row는 감사·reconciliation 보존 기간 뒤 bounded cleanup job으로 삭제하거나 archive하며, PENDING, CLAIMED, RETRYABLE_FAILURE, PARKED row는 자동 삭제하지 않는다.

30.4 Debezium Profile

Debezium Outbox Event Router integration은 선택 recipe다.

outbox ID를 message ID로 사용
aggregate ID를 Kafka key로 사용 가능
CDC engine 자체 운영은 이 모듈 범위 밖
polling relay와 CDC relay를 동시에 활성화하지 않음

31. Inbox와 Idempotent Consumer

31.1 테이블

create table messaging_inbox (
    consumer_name varchar(200) not null,
    message_id uuid not null,
    message_type varchar(240) not null,
    payload_hash varchar(128) not null,
    received_at timestamptz not null,
    processed_at timestamptz not null,
    expires_at timestamptz not null,
    primary key (consumer_name, message_id)
);

create index ix_messaging_inbox_expiry
    on messaging_inbox(expires_at);

31.2 처리 알고리즘

DB transaction 시작
→ Inbox INSERT
→ unique conflict이면 duplicate로 판정하고 business mutation 생략
→ 신규 row이면 business mutation 실행
→ transaction commit
→ commit 후 source ACK

business mutation이 실패하면 Inbox INSERT도 rollback되어야 한다.

31.3 Retention

Inbox retention
  >= broker retention
   + 최대 DLQ 보관 기간
   + 최대 replay/redrive 기간
   + 안전 여유

retention을 짧게 잡아 오래된 replay가 side effect를 다시 만드는 구성을 startup warning 또는 validation failure로 처리한다.

32. Claim Check

1 MiB를 넘는 payload는 기본적으로 object storage 또는 fileserver에 저장하고 reference를 메시지로 전달한다.

public record ClaimCheckReference(
        String store,
        String objectId,
        long size,
        String checksumAlgorithm,
        String checksum,
        String contentType,
        Instant expiresAt) {
}

규칙은 다음과 같다.

public signed URL을 메시지에 직접 넣지 않는다.
consumer가 자신의 service credential로 조회한다.
size·checksum·content type을 검증한다.
object retention은 message retention·retry·DLQ·redrive 기간보다 길어야 한다.
object 삭제는 모든 consumer 처리 완료를 알 수 없으므로 명시적 lifecycle 정책을 사용한다.
Claim Check fetch 실패는 PROCESSING_TRANSIENT 또는 permanent integrity failure로 분류한다.

33. Topology Manifest와 Provisioning

33.1 Manifest

public record TopologyManifest(
        List<DestinationTopology> destinations,
        List<RetryTopology> retries,
        List<DeadLetterTopology> deadLetters,
        TopologyManagementMode managementMode) {
}

public enum TopologyManagementMode {
    AUTO_CREATE_DEV,
    VALIDATE_ONLY,
    ADMIN_MANAGED
}

33.2 환경 정책

환경	정책
local/test	auto-create 허용
development	auto-create 선택
staging	IaC + validate-only
production	IaC + validate-only, destructive mutation 금지

33.3 검증 항목

Kafka:

partition count
replication factor
min.insync.replicas
retention
cleanup policy
transaction capability

RabbitMQ:

exchange type/durability
queue type/durability
binding
quorum policy
dead-letter strategy
overflow
consumer timeout

Pulsar·NATS는 adapter capability에 따라 동일한 logical manifest에 native properties를 추가한다.

34. Admin Plane

public interface MessagingAdminService {
    TopologyValidationReport validateTopology();
    ReplayPlan planReplay(ReplayRequest request);
    ReplayResult executeReplay(ApprovedReplayPlan plan);
    RedrivePlan planRedrive(RedriveRequest request);
    RedriveResult executeRedrive(ApprovedRedrivePlan plan);
}

파괴 작업은 별도 interface에 둔다.

public interface DestructiveMessagingAdmin {
    DestructiveResult resetOffset(ApprovedOffsetReset request);
    DestructiveResult purge(ApprovedPurge request);
    DestructiveResult deleteDestination(ApprovedDelete request);
}

34.1 Guard

app credential로 bean 미생성
별도 admin role
dry-run 기본
approval ID 필수
대상·범위·예상 message 수 표시
실행 전 topology version 재검증
audit event 저장
동일 approval의 중복 실행 방지

35. Pulsar Experimental Adapter

지원 범위는 다음과 같다.

typed publish/consume
Exclusive / Failover / Shared / Key_Shared
broker acknowledgement
schema integration
redelivery
DLQ policy
replay cursor
multi-topic transaction capability

Stable 승격 조건:

4.0 LTS와 4.2에서 Core Contract Suite 통과
Shared와 Key_Shared ordering 차이를 profile validator가 차단
transaction commit·abort·transactional ACK 장애 테스트 통과
native schema와 공통 Schema Registry 계약 정합성 확인
TLS·authentication·authorization production guard 통과

Pulsar 5 preview는 지원 기준으로 사용하지 않는다.

36. NATS JetStream Experimental Adapter

Core NATS pub/sub은 at-most-once이므로 reliability destination의 기본 transport로 사용하지 않는다.

JetStream 지원 범위:

stream publish + PubAck
Nats-Msg-Id deduplication
pull consumer
durable consumer
Explicit ACK
AckSync
Nak / NakWithDelay
AckWait / BackOff
MaxAckPending
MaxDeliver advisory
LimitsPolicy
WorkQueuePolicy
replay by sequence/time

NATS에는 Kafka·Pulsar형 transaction을 광고하지 않는다.

36.1 DLQ workflow

MaxDeliver 도달만으로 메시지가 자동 DLQ 이동됐다고 가정하지 않는다.

MaxDeliver advisory
→ parking workflow consumer
→ target publish + PubAck
→ 원본 terminal settlement 또는 admin mark

Stable 승격 조건:

dedupe window 경계 테스트
Ack()와 AckSync() 차이 검증
MaxDeliver advisory 기반 parking 테스트
stream failover와 redelivery contract 통과

37. Spring Cloud Stream Bridge

Spring Cloud Stream은 optional bridge다. Core contract를 대체하지 않는다.

public interface MessagingBindingBridge {
    void bindPublisher(DestinationName destination, String bindingName);
    void bindConsumer(DestinationName destination, String bindingName);
}

제약은 다음과 같다.

binder retry·DLQ 의미론을 Core 정책으로 자동 승격하지 않음
binder extended property는 M3 capability로 격리
Kafka transaction과 Rabbit routing 차이를 bridge가 숨기지 않음
Core PublishResult와 settlement evidence를 제공할 수 없는 binder 경로는 migration-only 등급

38. Security

38.1 Production 기본

TLS 필수
hostname/certificate validation 필수
broker 인증 필수
producer/consumer/admin identity 분리
destination-level 최소 권한
tenant namespace 분리
credential rotation 가능
secret header 금지
admin API 별도 credential

38.2 Credential Profile

public sealed interface BrokerCredentialProfile
        permits BrokerCredentialProfile.SaslScram,
                BrokerCredentialProfile.OAuth2,
                BrokerCredentialProfile.MutualTls,
                BrokerCredentialProfile.UsernamePassword,
                BrokerCredentialProfile.Nkey {
    String credentialId();
    record SaslScram(String credentialId) implements BrokerCredentialProfile {}
    record OAuth2(String credentialId) implements BrokerCredentialProfile {}
    record MutualTls(String credentialId) implements BrokerCredentialProfile {}
    record UsernamePassword(String credentialId) implements BrokerCredentialProfile {}
    record Nkey(String credentialId) implements BrokerCredentialProfile {}
}

credential은 secret provider에서 runtime에 가져온다. config 파일과 exception에 원문을 남기지 않는다.

38.3 ACL

Kafka topic/group/transactional ID 권한 분리
Rabbit vhost, exchange, queue configure/write/read 권한 분리
Pulsar tenant/namespace/topic 권한 분리
NATS account/subject publish·subscribe allowlist
Admin identity만 topology mutation 권한 보유

38.4 Message Security

payload-level encryption은 별도 codec profile
trace baggage allowlist
tenant context는 신뢰 경계에서 재검증
message header에 token·password·PII 원문 금지
schema parser에 크기·깊이 제한

39. Observability

OpenTelemetry messaging semantic convention은 exporter adapter로 사용하되 내부 public API에 convention attribute 이름을 고정하지 않는다.

39.1 Metric

영역	Metric
Producer	publish count, confirmed, rejected, ambiguous, confirmation latency
Consumer	received, processed, failed, processing latency
Settlement	ack, retry, reject, dead-letter, unknown, latency
Reliability	redelivery, retry, retry exhausted, DLQ, redrive
Backlog	consumer lag, queue depth, pending ACK
Runtime	connection, reconnect, rebalance, assignment
Schema	encode/decode failure, incompatibility
Resource	producer buffered bytes, in-flight, batch size, prefetch
Admin	replay, redrive, offset reset, purge, delete
Outbox	pending, leased, publish success, ambiguous, parked, age
Inbox	inserted, duplicate, cleanup

39.2 Low-cardinality tag

허용:

broker
destinationProfile
destinationTemplate
operation
messageType  // bounded catalog일 때만
consumerGroupProfile
outcome
failureCategory
retryStage
schemaCodec

금지:

messageId
partitionKey
orderingKey
payload
actual dynamic destination
full exception message
customerId
unbounded tenantId
credential

39.3 Trace

messaging.publish logical span
  └─ broker publish attempt span

messaging.consume process span
  ├─ handler span
  ├─ retry publish span
  ├─ DLQ publish span
  └─ settlement span

producer와 consumer는 비동기 경계이므로 consumer process span은 message creation context와 span link를 사용한다.

39.4 Logging

INFO  destinationProfile=order-events outcome=CONFIRMED attempt=1
WARN  failureCategory=PROCESSING_TRANSIENT retryStage=RETRY_DESTINATION attempt=3
ERROR failureCategory=AMBIGUOUS destinationProfile=payment-events

payload와 full headers는 기본 로깅하지 않는다.

40. Backpressure·Payload·Concurrency

40.1 공통 제한

max payload bytes
max header bytes
max batch count
max batch bytes
max producer in-flight
max consumer concurrent handlers
max processing time
max retry concurrent publish
max DLQ backlog alarm

40.2 Broker mapping

공통	Kafka	RabbitMQ	Pulsar	NATS
producer buffered bytes	`buffer.memory`	channel confirm window	pending messages	publish pending
max in-flight	producer config	confirm outstanding	pending queue	async publish pending
batch	`batch.size`	publisher batch	batching	publish batch
consumer concurrency	partitions	consumer count	subscription consumers	pull workers
consumer in-flight	poll/partition tracker	prefetch	receiver queue	MaxAckPending
backlog	lag	queue depth	backlog	stream pending

40.3 Overload 동작

producer buffer 포화 시 bounded wait 후 MessageBackpressureException
consumer handler queue 포화 시 partition/consumer pause
retry publish concurrency는 원본 consume concurrency와 분리
DLQ 장애가 지속되면 source consumer를 pause하고 alert
payload limit은 broker 전송 전에 local reject
batch bytes와 count 둘 다 제한

41. Lifecycle과 Graceful Shutdown

종료 순서는 다음과 같다.

신규 publish admission 차단
→ 신규 delivery handler 시작 차단
→ consumer pause
→ in-flight handler drain
→ 완료된 settlement 전송
→ producer confirm 대기
→ Outbox lease 반환
→ connection close

30초 기본 drain 이후 남은 작업은 다음처럼 처리한다.

unconfirmed publish: AMBIGUOUS 기록
unfinished handler: source 미settlement로 redelivery 허용
Outbox claimed row: lease 만료 후 재처리
manual settlement 미완료: warning + source redelivery

JVM shutdown hook만 신뢰하지 않고 Spring lifecycle phase를 사용한다.

42. Spring Boot Starter

42.1 Auto-configuration

MessagingCoreAutoConfiguration
MessagingSchemaAutoConfiguration
MessagingPolicyAutoConfiguration
KafkaMessagingAutoConfiguration
RabbitMessagingAutoConfiguration
MessagingReliabilityAutoConfiguration
MessagingAdminAutoConfiguration
MessagingObservabilityAutoConfiguration
MessagingSecurityAutoConfiguration

Experimental adapter는 별도 property와 classpath가 모두 있어야 활성화된다.

42.2 Configuration Properties

@ConfigurationProperties("backend.messaging")
public record MessagingProperties(
        Map<String, BrokerProperties> brokers,
        Map<String, DestinationProperties> destinations,
        MessagingLimitsProperties limits,
        MessagingSecurityProperties security,
        MessagingObservabilityProperties observability,
        MessagingAdminProperties admin) {
}

42.3 Actuator

/actuator/messaging
/actuator/messaging/topology
/actuator/messaging/outbox
/actuator/messaging/capabilities

Actuator는 payload, actual credential, 전체 dynamic topic name을 노출하지 않는다.

43. 테스트 전략

43.1 공통 Contract Suite

모든 adapter는 다음을 통과한다.

normal publish
publish reject
publish ambiguity
same messageId retry
normal consume
handler failure
settlement loss
redelivery
retry exhausted
DLQ confirmation
DLQ failure
redrive identity
ordering scope
payload limit
header policy
schema failure
credential failure
ACL failure
TLS failure
graceful shutdown
observability cardinality

43.2 Kafka Native Suite

acks=all
idempotent producer
leader failover
publish timeout ambiguity
rebalance
contiguous offset commit
pause/resume
retry topic reorder
read_committed
transaction commit/abort
producer fencing
seek/replay
Share Group acknowledgement experimental

43.3 Rabbit Native Suite

publisher confirm
mandatory return
unroutable
confirm timeout
channel close
manual ACK loss
requeue
prefetch
quorum minority loss
at-least-once DLX target outage
retry queue cycle
single active consumer order

43.4 Reliability Suite

DB commit 전 crash
DB commit 후 relay 전 crash
broker store 후 confirm loss
same messageId duplicate publish
consumer DB commit 후 ACK 전 crash
Inbox duplicate effect 차단
Outbox lease expiry
redrive duplicate request
Claim Check checksum mismatch

43.5 Pulsar·NATS Suite

Pulsar:

Shared / Key_Shared
transaction commit/abort
transactional ACK
schema compatibility
subscription redistribution

NATS:

PubAck
Nats-Msg-Id dedupe
Ack / AckSync
AckWait expiry
NakWithDelay
MaxDeliver advisory
stream failover

43.6 장애 도구

Testcontainers
Toxiproxy
process kill
broker node kill
network partition
latency / packet loss
PostgreSQL restart
credential rotation
TLS certificate rotation

44. 호환성 인증 매트릭스

대상	PR	Nightly	Release Gate
Kafka 4.2	Core subset	Full	필수
Kafka 4.3.x	Full	Full + chaos	필수
RabbitMQ 4.3.x	Full	Full + chaos	필수
PostgreSQL 16 Outbox/Inbox	Full	crash suite	필수
Pulsar 4.0 LTS	smoke	Full	Experimental 필수
Pulsar 4.2	smoke	Full	Experimental 필수
NATS 2.14.x	smoke	Full	Experimental 필수
Spring Framework 6.2 line	compile/test	Full	필수
Spring Framework 7.0 line	compile/test	Full	필수
TLS·ACL	subset	rotation	필수
Performance	없음	baseline	release candidate

45. 운영 설정 기본 정책

항목	기본
retry	비활성, destination별 명시
delivery	at-least-once Stable 기본
auto settlement	handler success 뒤
DLQ	production at-least-once destination에 필수
topology	production validate-only
TLS	production 필수
payload max	1 MiB
Java serialization	차단
message type	catalog 등록 필수
schema version	필수
producer confirm	Stable destination 필수
consumer auto commit	차단
admin	기본 비활성
experimental	기본 비활성

46. 단계별 출시

Phase 1: Foundation Alpha

core API
envelope
schema JSON
capability
publish result
consumer settlement
policy validation
testkit

완료 조건: broker 없이 Core Contract와 architecture test 통과.

Phase 2: Kafka Stable Beta

producer
consumer group
offset tracker
pause/resume
retry
transaction capability
replay

완료 조건: leader failover, rebalance, ambiguity, transaction suite 통과.

Phase 3: Rabbit Stable Beta

confirm + return
quorum queue
manual ACK
prefetch
retry queue
DLQ

완료 조건: confirm loss, ACK loss, unroutable, quorum failover, DLQ failure 통과.

Phase 4: Reliability RC

Outbox
Inbox
Idempotent Consumer
Claim Check

완료 조건: crash matrix에서 duplicate DB effect가 발생하지 않음.

Phase 5: Operations Release

security
observability
topology validation
admin replay/redrive
starter
documentation

Phase 6: Experimental Compatibility

Kafka Share Group
Pulsar
NATS JetStream
Spring Cloud Stream bridge

47. 비지원 범위

초기 Stable release는 다음을 지원하지 않는다.

generic EXACTLY_ONCE=true
global ordering
raw broker client injection
application runtime의 purge·delete·offset reset
automatic XA
DB+broker atomicity 광고
automatic infinite retry
source ACK 후 DLQ publish
one-shot payload의 publish retry
arbitrary Java object serialization
unlimited headers·payload·batch
production auto topology mutation
Core NATS를 durable delivery로 사용
Pulsar preview release 지원
Kafka Share Group을 ordered stream으로 사용
Rabbit queue를 application completion FIFO로 광고
replay를 일반 handler API에서 수행
signed public URL을 Claim Check payload로 사용

48. 구현 결정 원장

질문	확정 답변
Core가 Spring `Message<?>`를 노출하는가?	아니오
Core async 타입은 무엇인가?	`CompletionStage`
Reactive는 어디서 제공하는가?	facade/adapter module
기본 브로커는 무엇인가?	기본 하나를 정하지 않고 destination profile이 선택
Stable adapter는?	Kafka, RabbitMQ
Experimental adapter는?	Kafka Share Group, Pulsar, NATS
publish timeout은 성공·실패를 확정하는가?	아니오, `AMBIGUOUS` 가능
messageId는 retry에서 바뀌는가?	아니오
DLQ 이동 후 source ACK 순서는?	DLQ confirm 후 ACK
retry topic은 ordering을 보존하는가?	아니오
Outbox만으로 exactly-once인가?	아니오
Inbox transaction 범위는?	Inbox row + business DB effect
topology는 누가 생성하는가?	운영 IaC, dev/test만 auto-create
large payload는?	Claim Check
Raw bytes는?	M2
Java serialization은?	비지원
Admin credential은 app과 공유하는가?	아니오
Experimental 기능은 기본 활성화되는가?	아니오

49. 완료 정의

플랫폼은 다음 조건을 모두 만족할 때 Stable로 간주한다.

공개 계약

M1만으로 typed publish와 consume을 구현할 수 있다.
PublishResult가 evidence와 ambiguity를 표현한다.
handler success 이전 settlement가 구조적으로 불가능하다.
broker SDK 타입이 Core에 노출되지 않는다.

Kafka

idempotent producer와 acks=all guard가 동작한다.
leader failover에서 confirmed, rejected, ambiguous를 구분한다.
rebalance와 async handler에서 contiguous offset만 commit한다.
transaction capability의 범위를 Kafka 내부로 제한한다.

RabbitMQ

confirm과 routing outcome을 결합한다.
unroutable message를 confirmed success로 처리하지 않는다.
ACK 유실과 publish confirm 유실을 별도로 검증한다.
quorum queue와 DLQ 조건을 startup에서 검증한다.

Reliability

Outbox relay crash 후 같은 message ID로 재발행한다.
Inbox가 duplicate DB effect를 차단한다.
DLQ·redrive에서 identity와 schema를 유지한다.
Claim Check가 size·checksum·retention을 검증한다.

운영

topology drift를 startup 또는 admin report에서 발견한다.
app credential로 destructive operation이 불가능하다.
TLS·ACL·credential rotation suite가 통과한다.
metric cardinality와 secret leak 검사가 통과한다.
payload·in-flight·backlog 상한 안에서 부하 테스트가 통과한다.

문서

support matrix
configuration reference
delivery guarantees
retry and DLQ guide
outbox and inbox guide
broker operations runbook
migration guide
experimental feature policy

이 문서의 설계 결정은 구현 계획의 Global Constraints와 각 Task의 acceptance test에 그대로 반영한다.

73 KiB Raw Permalink Blame History Unescape Escape