# 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가 동일한 형식으로 답할 수 있어야 한다. ```text 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 목표 1. 일반 서비스가 브로커 종류를 몰라도 typed message를 publish·consume할 수 있게 한다. 2. 브로커별 보장 차이를 숨기지 않고 capability와 evidence로 노출한다. 3. publish 결과가 확정적이지 않으면 `AMBIGUOUS`로 표현한다. 4. consumer duplicate, ACK 유실, rebalance, failover를 정상적인 운영 시나리오로 다룬다. 5. retry·DLQ·redrive에서 logical `messageId`와 schema identity를 보존한다. 6. Kafka와 RabbitMQ를 Stable Adapter로 제공한다. 7. Pulsar와 NATS JetStream을 Experimental Adapter로 제공해 abstraction boundary를 검증한다. 8. Outbox·Inbox를 실행 가능한 Reliability Recipe로 제공한다. 9. 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 포함 범위 ```text 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 모듈 ```text Transactional Outbox Inbox Idempotent Consumer Claim Check Debezium Outbox Event Router integration ``` ### 4.3 별도 기술 모듈 ```text 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. 핵심 설계 원칙 1. **증거를 보장 이름보다 우선한다.** `BROKER_ACK`, `REPLICATION_OR_PERSISTENCE_ACK`, routing result를 별도 필드로 표현한다. 2. **모호성을 숨기지 않는다.** 전송 후 confirm을 받지 못하면 성공 또는 실패로 추정하지 않고 `AMBIGUOUS`를 반환한다. 3. **중복 전달을 기본 전제로 둔다.** consumer는 handler·DB·settlement 경계에서 duplicate-safe해야 한다. 4. **logical identity를 유지한다.** publish retry, redelivery, DLQ, redrive에서 `messageId`를 바꾸지 않는다. 5. **ordering scope를 제한한다.** 순서는 destination 전체가 아니라 partition·key·single consumer 같은 실제 범위로만 표현한다. 6. **공통 API가 최소공배수가 되지 않게 한다.** 공통 계약과 broker-native capability를 분리한다. 7. **정책 우회 경로를 만들지 않는다.** M2·M3도 TLS, credential, destination ACL, payload limit, trace, masking을 우회할 수 없다. 8. **운영 변경은 Admin Plane에서 수행한다.** 애플리케이션 runtime은 describe·validate만 수행한다. 9. **자동 retry는 opt-in이다.** 실패 분류, ordering, idempotency, deadline, broker capability가 모두 허용할 때만 수행한다. 10. **DLQ도 publish다.** DLQ publish 실패를 source success로 처리하지 않는다. 11. **Outbox는 duplicate를 제거하지 않는다.** stable message ID와 Inbox가 함께 있어야 effectively-once DB effect를 구성한다. 12. **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. 전체 아키텍처 ```text 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 흐름 ```text 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 흐름 ```text 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. 모듈 구조 ```text 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 의존 방향 ```text 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. 핵심 식별자와 타입 ```java 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 ```java public record MessageEnvelope( MessageId messageId, MessageType messageType, SchemaVersion schemaVersion, Instant producedAt, Optional occurredAt, ProducerId producer, Optional correlationId, Optional causationId, ContentType contentType, Optional partitionKey, Optional orderingKey, Optional 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 정책 ```java public final class MessageHeaders { private final Map values; public static MessageHeaders of(Map values) { return HeaderPolicy.validateAndCopy(values); } } ``` 예약 header는 platform만 쓸 수 있다. ```text 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는 다음과 같다. ```text Authorization Proxy-Authorization Cookie Set-Cookie access_token refresh_token api_key password client_secret ``` --- ## 12. 논리 Destination과 Profile ```java public record MessageDestination( DestinationName name, MessageType messageType, Class payloadType) { } ``` 애플리케이션은 `topic`, `exchange`, `queue`, `subject`가 아니라 logical destination을 사용한다. 물리 매핑은 `DestinationProfile`이 소유한다. ```yaml 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는 브로커가 모든 기능을 동일하게 지원한다고 가정하지 않는다. ```java 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을 만든다. ```java 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 ```java public interface MessageCodec { ContentType contentType(); EncodedMessage encode( MessageType type, SchemaVersion version, Object payload); T decode( MessageType type, SchemaVersion version, EncodedMessage encoded, Class payloadType); } public record EncodedMessage( byte[] bytes, ContentType contentType, Optional 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 ```text 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를 보유한다. ```text schemas//v1/valid.* schemas//v2/valid.* schemas//invalid/* ``` CI에서 구버전 fixture를 최신 consumer가 읽고, compatibility mode에 따라 신버전 fixture를 구버전 consumer가 읽는지 검증한다. --- ## 15. CloudEvents Profile CloudEvents는 Domain·Integration Event의 상호운용 profile로만 제공한다. Command와 Work Item에는 강제하지 않는다. ```java public interface CloudEventMapper { CloudEvent toCloudEvent(MessageEnvelope envelope, URI source); MessageEnvelope 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 공개 계약 ```java public interface MessagePublisher { CompletionStage publish( MessageDestination destination, MessageEnvelope message, PublishOptions options); } public record PublishOptions( Duration timeout, ConfirmationRequirement confirmation, Optional deduplication, Map brokerHints) { } ``` M1에서는 `brokerHints`가 비어 있어야 한다. M3 adapter만 typed hint를 추가할 수 있다. ### 16.1 Blocking·Reactive facade ```java public interface BlockingMessagePublisher { PublishResult publish( MessageDestination destination, MessageEnvelope message, PublishOptions options); } public interface ReactiveMessagePublisher { Mono publish( MessageDestination destination, MessageEnvelope message, PublishOptions options); } ``` Core 구현은 `CompletionStage`를 사용하고 Blocking·Reactive facade가 lifecycle과 cancellation을 변환한다. ### 16.2 Batch Publish ```java public interface BatchMessagePublisher { CompletionStage publish( List> requests, BatchPublishOptions options); } ``` Batch는 transaction이 아니다. 결과는 입력 index별 성공·실패·모호성을 유지한다. ```java public record BatchPublishResult( List items, Duration elapsed) { } ``` 자동 batch retry는 하지 않는다. 호출자가 duplicate-safe한 item만 같은 `messageId`로 다시 제출한다. --- ## 17. Publish 결과와 증거 ### 17.1 결과 모델 ```java public record PublishResult( PublishCompletion completion, PublishEvidence evidence, RoutingOutcome routingOutcome, Optional position, int attempts, Duration elapsed, Optional 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 ```java public interface BrokerPosition { String broker(); Map diagnosticAttributes(); } public record KafkaPosition( String topic, int partition, long offset) implements BrokerPosition { @Override public String broker() { return "kafka"; } @Override public Map 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 diagnosticAttributes() { return Map.of("exchange", exchange, "routingKey", routingKey, "sequence", Long.toString(sequence)); } } ``` Position은 진단·replay에 쓰며 공통 업무 로직의 분기 기준으로 사용하지 않는다. --- ## 18. Producer 상태 머신 ```text 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 공개 계약 ```java public interface MessageHandler { CompletionStage handle(MessageDelivery delivery); } public record MessageDelivery( MessageEnvelope 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 ```java public record DeliveryMetadata( DestinationName destination, int deliveryAttempt, boolean redelivered, Optional brokerPosition, Optional partitionOrQueue, Optional 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 ```java public interface ManualMessageHandler { CompletionStage handle( MessageDelivery delivery, SettlementController settlement); } public interface SettlementController { CompletionStage ack(); CompletionStage retry(Duration delay); CompletionStage deadLetter(FailureDescriptor failure); CompletionStage reject(FailureDescriptor failure); } ``` ### 20.1 Batch Consume ```java public interface BatchMessageHandler { CompletionStage handle(BatchMessageDelivery batch); } public record BatchMessageDelivery( List> deliveries, BatchDeliveryMetadata metadata) { } ``` Batch consume은 M2다. batch 전체 settlement가 가능한 broker에서는 전체 결과를 사용하고, 개별 settlement가 가능한 broker에서는 item 결과를 유지한다. strict ordering destination에서는 하나의 ordering unit을 넘는 batch를 허용하지 않는다. ### 20.2 Pause·Resume와 Delayed Publish ```java public interface PauseResumeController { CompletionStage pause(DestinationName destination, String scope); CompletionStage resume(DestinationName destination, String scope); } public interface DelayedMessagePublisher { CompletionStage publish( MessageDestination destination, MessageEnvelope 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 증거 ```text RECEIVED ↓ DECODING ├─→ SCHEMA_PARKING ↓ PROCESSING ├─→ RETRY_PENDING ├─→ DEAD_LETTER_PENDING ├─→ REJECTED ↓ HANDLER_SUCCEEDED ↓ SETTLEMENT_SENDING ├─→ SETTLED └─→ SETTLEMENT_UNKNOWN ``` Reliability module을 사용할 때만 다음 증거를 추가한다. ```text INBOX_RESERVED BUSINESS_TRANSACTION_COMMITTED ``` ### 21.1 Settlement 결과 ```java public record SettlementResult( SettlementCompletion completion, SettlementEvidence evidence, Optional failure) { } public enum SettlementCompletion { SETTLED, REJECTED, UNKNOWN } ``` `UNKNOWN`은 redelivery 가능성을 의미한다. 호출자는 이를 성공으로 간주하지 않는다. --- ## 22. 전달·처리·순서 보장 모델 ```java 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에 존재하지 않는다. ```java 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. 오류 모델 ```text 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 ```java public enum FailureCategory { TRANSIENT_INFRASTRUCTURE, THROTTLED, PROCESSING_TRANSIENT, PERMANENT_BUSINESS, POISON_MESSAGE, DESERIALIZATION, AUTHENTICATION, AUTHORIZATION, AMBIGUOUS, CONFIGURATION } ``` ### 23.2 Failure Descriptor ```java public record FailureDescriptor( FailureCategory category, String code, boolean retryable, String sanitizedMessage, Optional exceptionType) { } ``` Payload, full stack trace, credential, actual message key는 message header에 적재하지 않는다. Stack trace는 secure log storage에만 기록한다. --- ## 24. Retry Policy Engine ### 24.1 정책 모델 ```java public record RetryPolicy( RetryMode mode, int maxAttempts, Duration initialDelay, Duration maxDelay, double multiplier, boolean jitter, OrderingImpact orderingImpact, Set retryableCategories, Set nonRetryableCategories) { } public enum RetryMode { NONE, INLINE, BLOCKING, PAUSE_PARTITION, RETRY_DESTINATION, BROKER_DELAYED } public enum OrderingImpact { PRESERVE, ALLOW_REORDER } ``` ### 24.2 결정 입력 ```java public record RetryContext( DestinationProfile destination, DeliveryMetadata delivery, FailureDescriptor failure, MessagingCapabilities capabilities, boolean handlerMayHaveCommittedSideEffect) { } ``` ### 24.3 결정 결과 ```java 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 처리 순서 ```text 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 ```java public record DeadLetterMetadata( MessageId originalMessageId, DestinationName originalDestination, Optional originalPosition, Instant firstFailureAt, Instant lastFailureAt, int attempts, FailureCategory failureCategory, String failureCode, int redriveCount, Optional lastRedriveId) { } ``` 원본 payload와 schema identity를 유지한다. 전체 stack trace는 header에 넣지 않는다. ### 25.3 Parking 역직렬화 불가, schema 미지원, 관리자 판정이 필요한 메시지는 일반 DLQ와 분리된 parking destination에 저장한다. ```text schema-parking security-parking manual-review-parking ``` ### 25.4 Redrive ```java public record RedriveRequest( DestinationName sourceDlq, DestinationName target, Set 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은 다음을 강제한다. ```text enable.idempotence=true acks=all retries > 0 max.in.flight.requests.per.connection <= 5 ``` 플랫폼 timeout과 Kafka delivery timeout의 관계를 startup에서 검증한다. ```text 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는 다음을 사용한다. ```text enable.auto.commit=false isolation.level=read_committed // transactional destination profile ``` M1의 strict order path는 partition별 한 개의 in-flight handler를 기본으로 한다. ```text 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 ```java 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로 제공한다. ```java public interface KafkaTransactionalProcessor { CompletionStage process( KafkaTransactionalDelivery delivery, KafkaTransactionalPublisher publisher); } ``` 보장 범위는 Kafka input offset과 Kafka output record다. 외부 DB·HTTP side effect는 포함하지 않는다. ### 26.8 Replay Replay는 M2/Admin에서만 제공한다. ```text seek by offset seek by timestamp replay to isolated consumer group replay to new destination ``` 기존 production consumer group offset reset은 M4 위험 작업이다. ### 26.9 Tombstone ```java public interface KafkaTombstonePublisher { CompletionStage publishTombstone( DestinationName destination, String key, MessageHeaders headers); } ``` 일반 `MessageEnvelope`의 null payload로 대체하지 않는다. --- ## 27. Kafka Share Group Experimental Adapter Kafka 4.2+ Share Group은 record 단위 acknowledgement와 delivery attempt counting을 제공하지만 ordered stream abstraction으로 사용하지 않는다. ```java public interface KafkaShareWorkQueueCapability { void register( MessageDestination destination, MessageHandler 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는 다음을 강제한다. ```text publisher-confirm-type=correlated publisher-returns=true mandatory=true ``` confirm과 return을 correlation ID로 결합한다. ```java public record RabbitPublishOutcome( RabbitConfirmOutcome confirm, RoutingOutcome routing, Optional replyCode, Optional 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 기본값 ```text durable=true auto-delete=false exclusive=false queue-type=quorum ``` classic queue는 명시적 low-durability profile에서만 허용한다. ### 28.5 Ordering Rabbit queue는 enqueue FIFO를 출발점으로 하지만 다음 조건에서 처리 완료 순서는 달라질 수 있다. ```text 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 두 모드를 제공한다. 1. **Platform-managed DLQ** - target publish confirm 후 source ACK - broker-neutral 기본 2. **Native quorum at-least-once dead-lettering** - M3 capability - 필요한 queue policy를 startup에서 검증 기본 dead-letter strategy가 at-most-once인 topology를 stronger guarantee로 표현하지 않는다. ### 28.7 Retry ```text 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 ```java public interface ReliableMessagePublisher { void addToOutbox( MessageDestination destination, MessageEnvelope message); } public interface IdempotentMessageHandler { CompletionStage handleOnce( String consumerName, MessageDelivery delivery, TransactionalMessageAction action); } ``` Reliability module은 Core의 publish·delivery 계약을 바꾸지 않는다. DB transaction과 message identity를 결합한다. --- ## 30. Transactional Outbox ### 30.1 테이블 ```sql 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 상태 ```text PENDING CLAIMED PUBLISHED RETRYABLE_FAILURE PARKED ``` ### 30.3 Relay 알고리즘 ```text 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 테이블 ```sql 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 처리 알고리즘 ```text 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 ```text 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를 메시지로 전달한다. ```java 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 ```java public record TopologyManifest( List destinations, List retries, List 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: ```text partition count replication factor min.insync.replicas retention cleanup policy transaction capability ``` RabbitMQ: ```text 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 ```java public interface MessagingAdminService { TopologyValidationReport validateTopology(); ReplayPlan planReplay(ReplayRequest request); ReplayResult executeReplay(ApprovedReplayPlan plan); RedrivePlan planRedrive(RedriveRequest request); RedriveResult executeRedrive(ApprovedRedrivePlan plan); } ``` 파괴 작업은 별도 interface에 둔다. ```java 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 지원 범위는 다음과 같다. ```text 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 지원 범위: ```text 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 이동됐다고 가정하지 않는다. ```text 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를 대체하지 않는다. ```java 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 기본 ```text TLS 필수 hostname/certificate validation 필수 broker 인증 필수 producer/consumer/admin identity 분리 destination-level 최소 권한 tenant namespace 분리 credential rotation 가능 secret header 금지 admin API 별도 credential ``` ### 38.2 Credential Profile ```java 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 허용: ```text broker destinationProfile destinationTemplate operation messageType // bounded catalog일 때만 consumerGroupProfile outcome failureCategory retryStage schemaCodec ``` 금지: ```text messageId partitionKey orderingKey payload actual dynamic destination full exception message customerId unbounded tenantId credential ``` ### 39.3 Trace ```text 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 ```text 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 공통 제한 ```text 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 종료 순서는 다음과 같다. ```text 신규 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 ```text MessagingCoreAutoConfiguration MessagingSchemaAutoConfiguration MessagingPolicyAutoConfiguration KafkaMessagingAutoConfiguration RabbitMessagingAutoConfiguration MessagingReliabilityAutoConfiguration MessagingAdminAutoConfiguration MessagingObservabilityAutoConfiguration MessagingSecurityAutoConfiguration ``` Experimental adapter는 별도 property와 classpath가 모두 있어야 활성화된다. ### 42.2 Configuration Properties ```java @ConfigurationProperties("backend.messaging") public record MessagingProperties( Map brokers, Map destinations, MessagingLimitsProperties limits, MessagingSecurityProperties security, MessagingObservabilityProperties observability, MessagingAdminProperties admin) { } ``` ### 42.3 Actuator ```text /actuator/messaging /actuator/messaging/topology /actuator/messaging/outbox /actuator/messaging/capabilities ``` Actuator는 payload, actual credential, 전체 dynamic topic name을 노출하지 않는다. --- ## 43. 테스트 전략 ### 43.1 공통 Contract Suite 모든 adapter는 다음을 통과한다. ```text 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 ```text 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 ```text 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 ```text 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: ```text Shared / Key_Shared transaction commit/abort transactional ACK schema compatibility subscription redistribution ``` NATS: ```text PubAck Nats-Msg-Id dedupe Ack / AckSync AckWait expiry NakWithDelay MaxDeliver advisory stream failover ``` ### 43.6 장애 도구 ```text 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 ```text core API envelope schema JSON capability publish result consumer settlement policy validation testkit ``` 완료 조건: broker 없이 Core Contract와 architecture test 통과. ### Phase 2: Kafka Stable Beta ```text producer consumer group offset tracker pause/resume retry transaction capability replay ``` 완료 조건: leader failover, rebalance, ambiguity, transaction suite 통과. ### Phase 3: Rabbit Stable Beta ```text confirm + return quorum queue manual ACK prefetch retry queue DLQ ``` 완료 조건: confirm loss, ACK loss, unroutable, quorum failover, DLQ failure 통과. ### Phase 4: Reliability RC ```text Outbox Inbox Idempotent Consumer Claim Check ``` 완료 조건: crash matrix에서 duplicate DB effect가 발생하지 않음. ### Phase 5: Operations Release ```text security observability topology validation admin replay/redrive starter documentation ``` ### Phase 6: Experimental Compatibility ```text 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에 그대로 반영한다.