1547 lines
93 KiB
Markdown
1547 lines
93 KiB
Markdown
# Messaging 모듈 상세 코드·아키텍처 리뷰
|
||
|
||
- 기준 일자: 2026-08-14
|
||
- 기준 Git HEAD: `c3043e530a604315c4df341b87b5470c7617ea03`
|
||
- `src/messaging` source snapshot: `20664539b0609c6c413759e2b2945bf421c10de7`
|
||
- 범위: 신규 `src/messaging/*` 24개 Gradle leaf, 기존 `application-core`/`adapter:outbound:messaging`,
|
||
`app-bootstrap`, architecture registry, `docs/messaging`, 관련 테스트와 CI
|
||
- 규모: 신규 production Java 320개/22,338 LOC, test Java 87개/13,319 LOC
|
||
- 판정: **CHANGES REQUIRED — 신규 messaging platform을 runtime-ready Stable로 사용하면 안 됨**
|
||
- 변경 범위: 이 리뷰 문서만 추가했으며 production/test 코드는 수정하지 않았다.
|
||
|
||
> 이전 리뷰 뒤 merge가 완료됐다는 요청에 따라 과거 작업 결과를 재사용하지 않고 최종 HEAD의 파일,
|
||
> registry, runtime membership, 테스트 산출물을 다시 대조했다. 현재 애플리케이션은 기존 messaging
|
||
> adapter만 사용하며 신규 24개 leaf는 모두 `runtime_memberships: []`이다. 따라서 아래 결함은 현재
|
||
> bootstrap이 곧바로 장애 난다는 뜻이 아니라, 신규 starter를 소비하거나 cutover하는 순간 드러나는
|
||
> release blocker다.
|
||
|
||
## 1. 최종 결론
|
||
|
||
신규 messaging 코드는 API 타입 수나 단위 테스트 수가 부족해서 문제가 아니다. `PublishResult`의
|
||
`CONFIRMED`/`REJECTED`/`AMBIGUOUS` 구분, sealed handler result, immutable envelope, broker profile,
|
||
outbox/inbox, retry, security, observability, admin까지 필요한 개념은 넓게 갖췄다. 실제로 이번 fresh
|
||
실행에서 신규 600개와 기존 adapter 81개 테스트가 failure/skip 없이 통과했다.
|
||
|
||
문제는 이 조각들이 하나의 production 실행 경로로 조립되지 않았고, 가장 중요한 신뢰성 계약 몇 개가
|
||
기본/public 경로에서 깨진다는 점이다.
|
||
|
||
1. `OutboxRepository.append(record)`와 `InboxRepository.reserve(...)`는 문서상 caller transaction에
|
||
참여해야 하지만 실제 JDBC 구현은 raw 새 connection을 연다. 업무 rollback 뒤 ghost event를
|
||
발행하거나, inbox 예약만 commit되어 재전달의 업무 효과를 영구 유실할 수 있다.
|
||
2. outbox lease에는 owner/fencing token이 없다. lease가 만료되어 새 worker가 처리한 뒤 늦은 기존
|
||
worker가 결과를 덮어쓸 수 있다.
|
||
3. 신규 starter에는 production `MessagePublisher`, destination router, broker runtime factory,
|
||
consumer outcome pipeline이 없다. 반면 Kafka와 Rabbit을 한 starter가 동시에 끌어온다.
|
||
4. Kafka consumer에는 poll batch 일부를 건너뛰는 backpressure 처리, executor rejection 누수,
|
||
rebalance 뒤 stale settlement, commit 실패 시 로컬 watermark 선반영 문제가 있다.
|
||
5. Rabbit publisher에는 `multiple` confirm, 자동 timeout, synchronous send failure 정리, return
|
||
correlation을 책임지는 production channel bridge가 없고, consumer는 handler/settlement 예외까지
|
||
deserialization 실패로 간주해 discard한다.
|
||
6. 코드의 `backend.messaging`, 기존 runtime의 `app.messaging`, 문서의 bare `messaging` 세 설정
|
||
namespace가 서로 다르며, 문서에 있는 destination/broker/security 설정은 binder에 존재하지 않는다.
|
||
7. Stable/실 브로커 장애 커버리지는 실행 evidence가 아니라 hard-coded matrix가 스스로 주장한다.
|
||
Kafka 인증 버전도 선언한 4.2/4.3과 실제 컨테이너 4.1.0이 다르다.
|
||
8. 기존 runtime과 신규 platform 사이에 semantic bridge와 단일 publication authority 전환 계획이
|
||
없다. 두 모델의 `FAILED` 의미도 다르므로 단순 enum 매핑은 안전하지 않다.
|
||
|
||
따라서 첫 구현 목표는 폴더 이동이나 패턴 추가가 아니다. **데이터 원자성 → lease fencing → broker
|
||
settlement 정확성 → 실제 runtime composition → 단일 cutover authority → release evidence** 순서로
|
||
고쳐야 한다. P0와 P1이 닫히기 전 신규 platform은 `Contract-only/Build-only`로 표시하고 production
|
||
runtime membership을 추가하지 않는 것이 안전하다.
|
||
|
||
## 2. 검토 범위와 증거 경계
|
||
|
||
### 2.1 신규 모듈 구성
|
||
|
||
| family | leaf | 역할 |
|
||
|---|---|---|
|
||
| API | core-api, schema-api, reliability-api | envelope/publish/consume/schema/outbox/inbox 계약 |
|
||
| codec | schema-json, schema-avro, schema-protobuf, cloudevents | wire encoding과 compatibility |
|
||
| runtime policy/SPI | policy, transport-spi | profile/retry/admission/runtime lease |
|
||
| stable broker | kafka, rabbit | native transport/consumer/admin capability |
|
||
| reliability | outbox-jpa, inbox-jpa, claim-check | PostgreSQL JDBC outbox/inbox와 payload offload |
|
||
| ops | observability, security, admin-api, admin-runtime | metric/trace/credential/operator 기능 |
|
||
| experimental | kafka-share, pulsar, nats, spring-cloud-stream-bridge | opt-in adapter/bridge |
|
||
| Spring/test | spring-boot-starter, testkit | auto-configuration/facade/contract evidence |
|
||
|
||
`src/config/architecture/modules.json`에는 기존 19개와 신규 24개, 총 43개 leaf가 등록되어 있다.
|
||
`src/settings.gradle:37`도 43개를 기대한다. 그러나 root `AGENTS.md:52,105,185`와
|
||
`CLAUDE.md:24,46`는 아직 정확히 19개라고 선언한다. 신규 24개는 모두 runtime membership이 비어 있고
|
||
`src/messaging/CLAUDE.md`도 없다.
|
||
|
||
### 2.2 깊게 따라간 실행 흐름
|
||
|
||
```text
|
||
publish API
|
||
-> destination/profile/schema/security/admission
|
||
-> runtime generation lease
|
||
-> Kafka/Rabbit transport
|
||
-> broker evidence -> PublishResult
|
||
-> outbox state transition/retry
|
||
|
||
broker delivery
|
||
-> wire mapper
|
||
-> MessageHandler/HandleResult
|
||
-> retry/DLQ policy
|
||
-> broker settlement
|
||
-> transactional inbox + business effect
|
||
```
|
||
|
||
이상적인 흐름과 실제 production reference를 대조했다. 실제로는 위 중앙 publish/consumer pipeline이
|
||
존재하지 않고, broker adapter와 각 policy 객체가 독립 조각으로 남아 있다.
|
||
|
||
### 2.3 이 리뷰가 승인하지 않는 범위
|
||
|
||
- 신규 platform이 `app-bootstrap` full context에서 실제 publish/consume한다는 주장
|
||
- Kafka 4.2/4.3 호환성, Rabbit의 5개 failure scenario 전체 커버리지
|
||
- process kill/restart, rebalance 경쟁, connection loss를 포함한 다중 replica 안정성
|
||
- 대량 payload/header, 고 cardinality diagnostics, credential rotation 부하의 운영 한계
|
||
- 기존 runtime에서 신규 platform으로 rolling cutover/rollback할 수 있다는 주장
|
||
|
||
로컬 Docker 환경에서는 현재 존재하는 Kafka/Rabbit/PostgreSQL IT가 모두 실행됐지만, 존재하지 않는
|
||
시나리오나 다른 broker version을 그 결과로 추론하지 않았다.
|
||
|
||
## 3. 유지할 설계
|
||
|
||
다음 방향은 리팩터링하면서 보존한다.
|
||
|
||
- `PublishResult`가 성공/거절/모호함과 transmission/confirmation/routing evidence를 분리한다.
|
||
- 공통 API에서 `EXACTLY_ONCE`와 global ordering을 약속하지 않는다.
|
||
- `HandleResult`, `RetryDecision`의 sealed hierarchy는 exhaustive policy 처리를 돕는다.
|
||
- `MessageEnvelope.withPayload`가 identity를 보존하고 encoded/reliability record가 mutable bytes를
|
||
방어 복사한다.
|
||
- core API가 framework/broker dependency를 갖지 않고 transport SPI가 broker strategy 경계를 둔다.
|
||
- Rabbit confirm과 mandatory return을 서로 다른 증거로 모델링하려는 방향은 맞다.
|
||
- Kafka contiguous watermark, partition pause/seek, runtime generation lease라는 핵심 개념은 맞다.
|
||
- stable/experimental module을 분리하고 optional codec dependency를 별도 leaf로 둔 선택은 유지할 가치가
|
||
있다.
|
||
- 기존 application-owned port → outbound adapter → bootstrap composition 의존 방향은 canonical
|
||
boundary로 계속 사용해야 한다.
|
||
|
||
## 4. 우선순위 요약
|
||
|
||
| ID | 우선순위 | 심각도 | 주제 | 완료 조건 요약 |
|
||
|---|---|---|---|---|
|
||
| MSG-001 | P0 | Critical | outbox/inbox transaction 원자성 위반 | public port 경로에서 업무 row와 함께 commit/rollback |
|
||
| MSG-002 | P0 | Critical | outbox lease fencing 부재 | stale worker의 모든 terminal update가 DB 조건으로 거절 |
|
||
| MSG-003 | P0 | Critical | production publisher/router/consumer pipeline 미조립 | starter full context가 fake 없이 실제 transport까지 연결 |
|
||
| MSG-004 | P0 | Critical | Kafka dispatch/rebalance/commit settlement 경쟁 | batch/reject/revoke/commit-failure에서 skip·조기 commit 없음 |
|
||
| MSG-005 | P0 | Critical | Rabbit confirm/return/consumer 오류 처리 결함 | multiple/timeout/return/handler failure가 단일 state machine으로 처리 |
|
||
| MSG-006 | P1 | High | retry budget/backoff가 relay에 미연결 | attempt cap과 next-at이 durable하며 outage 중 hot loop 없음 |
|
||
| MSG-007 | P1 | High | starter classpath와 auto-config 설계 불완전 | core/Kafka/Rabbit/reliability/admin 선택형 starter |
|
||
| MSG-008 | P1 | High | 설정 namespace/binder/validator 단절 | `app.messaging` 하나로 문서 YAML 전체 fail-closed binding |
|
||
| MSG-009 | P1 | High | batch timeout/stop 계약 미구현 | batch deadline과 비동기 rejection 이후 미제출 항목이 명시됨 |
|
||
| MSG-010 | P1 | High | runtime drain/backpressure permit 결함 | generation별 deadline + once-only permit + lifecycle 연결 |
|
||
| MSG-011 | P1 | High | credential rotation 경쟁 | generation lease가 끝난 뒤에만 이전 secret clear |
|
||
| MSG-012 | P1 | High | wire identifier/header validation 부족 | CRLF/NUL/Unicode/size/reserved bypass가 broker 전 거절 |
|
||
| MSG-013 | P1 | High | observability cardinality/secret 경계 우회 | arbitrary diagnostics가 metric tag가 되지 않음 |
|
||
| MSG-014 | P1 | High | Stable certification이 실행 evidence와 분리 | fail-closed broker/version/scenario artifact gate |
|
||
| MSG-015 | P1 | High | legacy/new semantic bridge와 cutover 권한 부재 | 단일 writer/relay + outcome/wire compatibility + rollback |
|
||
| MSG-016 | P1 | Critical | envelope/header/outbox/CDC canonical 정보 유실·위조 | broker/DB/CDC round-trip이 동일 canonical envelope 보존 |
|
||
| MSG-017 | P1 | High | PublishOptions/capability/result 계약 미적용 | option별 deadline/지원 여부와 result truth table 강제 |
|
||
| MSG-018 | P1 | Critical | Kafka transaction callback이 transaction 밖에서 실행 | handler/output/offset이 실제 한 Kafka transaction 안에 있음 |
|
||
| MSG-019 | P2 | Medium | public/vendor surface와 Gradle exposure 불일치 | external compile fixture와 public API allowlist 통과 |
|
||
| MSG-020 | P2 | Medium | codec가 제한 검사 전에 전체 할당 | bounded stream/parser로 max+1에서 중단 |
|
||
| MSG-021 | P2 | High | admin idempotency가 process-local/선점-only | durable state/fingerprint/lease 기반 one-shot execution |
|
||
| MSG-022 | P2 | Medium | experimental adapter의 unknown failure 오분류 | typed pre-send만 REJECTED, 나머지는 보수적 AMBIGUOUS |
|
||
| MSG-023 | P2 | Medium | module 명칭·폴더·정본 정책 drift | JDBC/PostgreSQL 명칭, family policy, registry-derived count |
|
||
| MSG-024 | P1 | High | legacy runtime disabled/retry/payload/log 안전성 | disabled broker가 row를 소진하지 않고 wire/log bound 강제 |
|
||
| MSG-025 | P3 | Low | runbook/outbox reclaim 문서 drift | 실제 env/class/expired IN_FLIGHT predicate와 동기화 |
|
||
|
||
## 5. 상세 발견 사항과 구현 명세
|
||
|
||
### MSG-001 — JDBC outbox/inbox의 기본 port 경로가 caller transaction을 벗어난다
|
||
|
||
**근거**
|
||
|
||
- `OutboxRepository.java:12-22`는 `append`가 caller business transaction 안에서 실행돼야 한다고
|
||
명시한다.
|
||
- `JdbcOutboxRepository.java:91-109`에는 caller `Connection`을 받는 안전한 overload가 있지만,
|
||
실제 interface override는 `112-119`에서 `withConnection`을 호출하고 `253-259`에서 raw
|
||
`dataSource.getConnection()`을 열고 닫는다.
|
||
- `InboxRepository.java:9-25`도 reservation과 side effect가 같은 transaction이어야 한다고 명시한다.
|
||
- `IdempotentConsumer.java:55-60`은 transaction runner 안에서 interface 메서드
|
||
`inbox.reserve(...)`를 호출하지만, `JdbcInboxRepository.java:76-83`은 별도 raw connection을 연다.
|
||
- `OutboxPostgresIT.java:152-171`과 `InboxPostgresIT.java:109-124`의 rollback 검증은 안전한
|
||
`Connection` overload를 직접 호출한다. production/public port 경로를 검증하지 않는다.
|
||
|
||
**실패 시나리오**
|
||
|
||
```text
|
||
Inbox
|
||
T1: reserve()가 별도 connection에서 auto-commit
|
||
T2: business side effect 실행 후 rollback
|
||
redelivery: 이미 inbox row가 있으므로 duplicate 처리
|
||
결과: 업무 효과 영구 유실
|
||
|
||
Outbox
|
||
T1: business row 변경
|
||
T2: append()가 별도 connection에서 auto-commit
|
||
T1 rollback
|
||
relay: rollback된 업무의 event를 발행
|
||
결과: ghost publication
|
||
```
|
||
|
||
Spring transaction 안에서 raw Hikari `DataSource#getConnection()`을 부르는 것만으로 같은 resource에
|
||
자동 참여하지 않는다. 외부에서 `TransactionAwareDataSourceProxy`를 우연히 씌웠을 때만 동작하는 설계는
|
||
port 계약이 아니다.
|
||
|
||
**구현 결정: Unit of Work + transaction-aware adapter**
|
||
|
||
1. Spring JDBC 구현은 `JdbcTemplate`/`NamedParameterJdbcTemplate`를 사용하거나
|
||
`DataSourceUtils.getConnection/releaseConnection`으로 transaction-bound connection을 얻는다.
|
||
2. outbox append/inbox reserve는 active, non-read-only transaction과 같은 `DataSource` resource가
|
||
없으면 `OUTBOX_TRANSACTION_REQUIRED`/`INBOX_TRANSACTION_REQUIRED`로 fail-fast한다.
|
||
3. `Connection`을 reliability API에 노출하지 않는다. public overload는 제거하거나 adapter 내부
|
||
package-private helper로 낮춘다.
|
||
4. `IdempotentConsumer.TransactionRunner`는 임의 lambda가 아니라 application transaction port 또는
|
||
Spring adapter의 `TransactionTemplate`로 구성하고, repository와 동일 transaction manager/data source를
|
||
startup에서 검증한다.
|
||
5. outbox writer를 application use case transaction의 마지막 단계에 두되, transaction synchronization
|
||
`afterCommit`으로 옮기지 않는다. after-commit publish는 원자성을 다시 잃는다.
|
||
|
||
권장 port는 JDBC 타입을 넣는 대신 원자성 요구를 표현한다.
|
||
|
||
```java
|
||
interface TransactionalOutbox {
|
||
void append(OutboxRecord record); // active application UnitOfWork required
|
||
}
|
||
|
||
interface TransactionalInbox {
|
||
Reservation reserve(MessageId messageId, ConsumerId consumerId, Instant now);
|
||
}
|
||
```
|
||
|
||
**필수 테스트**
|
||
|
||
- `TransactionTemplate` 안에서 interface `append(record)`만 호출하고 business insert와 함께 rollback;
|
||
둘 다 없어야 한다.
|
||
- interface `reserve(...)`와 business insert 후 handler exception; 둘 다 rollback되고 재시도에서 handler가
|
||
실행돼야 한다.
|
||
- 정상 commit, transaction 없음, read-only transaction, 다른 `DataSource` transaction을 각각 검증한다.
|
||
- auto-configuration이 repository/runner를 서로 다른 transaction manager로 조립하면 context가 실패해야
|
||
한다.
|
||
|
||
**완료 조건**
|
||
|
||
테스트가 안전한 overload를 직접 호출하지 않고 production port와 실제 Spring transaction composition을
|
||
통과해야 한다.
|
||
|
||
### MSG-002 — outbox lease에 owner/fencing token이 없어 stale worker가 최신 결과를 덮어쓴다
|
||
|
||
**근거**
|
||
|
||
- migration `V1__messaging_outbox.sql:6-25`에는 `lease_expires_at`만 있고 owner/token/version이 없다.
|
||
- `JdbcOutboxRepository.java:48-69`는 만료된 `IN_FLIGHT`를 재claim하면서 status와 expiry만 갱신한다.
|
||
- `markPublished`, `markAmbiguous`, `markFailed`는 `143-171`에서 모두 `WHERE message_id = ?`만
|
||
사용한다. `releaseLease`도 owner/token 조건이 없다.
|
||
|
||
**실패 시나리오**
|
||
|
||
```text
|
||
relay A: row claim(token 없음), publish 대기
|
||
lease expiry
|
||
relay B: 같은 row reclaim, broker confirm, PUBLISHED 기록
|
||
relay A: 늦은 timeout/exception, AMBIGUOUS 또는 FAILED 기록
|
||
결과: 확정 발행 row가 재시도되거나 terminal 상태가 잘못 회귀
|
||
```
|
||
|
||
lease duration을 publish timeout보다 길게 검증하는 것은 발생 확률을 줄일 뿐 process pause, GC, broker
|
||
latency, scheduler stall을 데이터 정합성 제약으로 바꾸지 못한다.
|
||
|
||
**구현 결정: Lease + Fencing Token state machine**
|
||
|
||
V2 migration으로 다음을 추가한다.
|
||
|
||
```sql
|
||
ALTER TABLE messaging_outbox
|
||
ADD COLUMN lease_owner VARCHAR(160),
|
||
ADD COLUMN lease_token BIGINT NOT NULL DEFAULT 0,
|
||
ADD COLUMN next_attempt_at TIMESTAMPTZ;
|
||
```
|
||
|
||
claim은 token을 증가시키고 lease identity를 반환한다.
|
||
|
||
```sql
|
||
UPDATE messaging_outbox o
|
||
SET status = 'IN_FLIGHT',
|
||
lease_owner = :owner,
|
||
lease_token = o.lease_token + 1,
|
||
lease_expires_at = :expires
|
||
FROM claimable c
|
||
WHERE o.message_id = c.message_id
|
||
RETURNING ..., o.lease_owner, o.lease_token;
|
||
```
|
||
|
||
port도 message ID가 아니라 lease를 terminal command에 넘긴다.
|
||
|
||
```java
|
||
record OutboxLease(OutboxRecord record, String owner, long token, Instant expiresAt) {}
|
||
|
||
OutboxTransitionResult markPublished(OutboxLease lease, Instant now);
|
||
OutboxTransitionResult markAmbiguous(OutboxLease lease, FailureCode code, Instant now);
|
||
```
|
||
|
||
모든 transition은 다음 predicate와 update count 1을 요구한다.
|
||
|
||
```sql
|
||
WHERE message_id = :id
|
||
AND status = 'IN_FLIGHT'
|
||
AND lease_owner = :owner
|
||
AND lease_token = :token
|
||
```
|
||
|
||
0 rows면 성공으로 삼키지 말고 `STALE_LEASE` outcome/metric으로 기록한다. status transition 표를 한
|
||
`OutboxStateMachine`에 두고 임의 SQL이 terminal state를 직접 바꾸지 못하게 한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- A claim → expiry → B claim/publish → A ambiguous/failed/release가 모두 0-row stale result.
|
||
- 같은 worker의 duplicate terminal call도 두 번째는 거절.
|
||
- 두 DB connection이 동시에 claim할 때 row 집합과 token이 겹치지 않음.
|
||
- process kill 뒤 expiry reclaim은 같은 `messageId`, 증가한 token으로 성공.
|
||
|
||
### MSG-003 — 신규 platform에는 production publisher/router와 consumer outcome pipeline이 없다
|
||
|
||
**근거**
|
||
|
||
- `messaging-core-api/.../publish/MessagePublisher.java:13-25`에 application-facing contract가 있지만
|
||
신규 production source에 구현이나 `@Bean MessagePublisher`가 없다.
|
||
- Kafka/Rabbit은 `MessagingTransport`만 구현한다.
|
||
- `MessagingCoreAutoConfiguration.java:104-107,176-203`은 publisher가 이미 있다고 가정해
|
||
`DeadLetterOrchestrator`, blocking/reactive/batch facade를 만든다.
|
||
- broker auto-config 문서도 producer/consumer를 만들지 않는다고 명시한다
|
||
(`KafkaMessagingAutoConfiguration.java:22-24`, `RabbitMessagingAutoConfiguration.java:19-21`).
|
||
- `MessagingAdmissionController`, `BackpressureController`, `MessagingRuntimeRegistry`, security validator,
|
||
observation은 실제 publish 경로에서 함께 호출되지 않는다.
|
||
- `MessageHandler`/`HandleResult`를 broker settlement/retry/DLQ로 변환하는 production orchestrator도 없다.
|
||
|
||
**영향**
|
||
|
||
starter에 application fake publisher를 넣으면 context 조각 테스트는 통과하지만 destination resolution,
|
||
encoding, admission, security, runtime lease, broker selection, timeout, evidence normalization이 실행되지
|
||
않는다. consumer도 handler가 반환한 retry/discard/success를 공통 정책과 정확히 연결하지 못한다.
|
||
|
||
**구현 결정: Facade + Pipeline/Decorator + Strategy**
|
||
|
||
새 `messaging-runtime-core` leaf에 중앙 orchestration을 둔다.
|
||
|
||
```text
|
||
DefaultMessagePublisher
|
||
-> DestinationRegistry.require(name)
|
||
-> CompositeProfileValidator / capability compiler
|
||
-> MessageAccessPolicy + MessageSecurityValidator
|
||
-> MessageCodecRegistry + bounded encode / ClaimCheck
|
||
-> AdmissionPermit
|
||
-> MessagingRuntimeLease
|
||
-> MessagingTransport.publish # broker Strategy
|
||
-> deadline/evidence normalization
|
||
-> observation
|
||
-> finally permit.close + lease.close
|
||
```
|
||
|
||
consumer counterpart는 다음 단일 흐름으로 둔다.
|
||
|
||
```text
|
||
BrokerDeliveryMapper
|
||
-> DefaultDeliveryProcessor
|
||
-> MessageHandler.handle
|
||
-> HandleResultVisitor
|
||
-> RetryDecisionEngine
|
||
-> DLQ publish confirmed
|
||
-> exactly one source settlement
|
||
```
|
||
|
||
- cross-cutting 단계를 자유로운 interceptor map으로 만들지 말고 순서가 고정된 typed decorator/list로 둔다.
|
||
- broker별 차이는 `MessagingTransport`와 `BrokerRuntimeFactory` Strategy에만 둔다.
|
||
- runtime factory는 profile 하나에서 producer/consumer/admin/security/lifecycle을 완결되게 만드는 Abstract
|
||
Factory다. 부분 bean graph는 startup에서 거절한다.
|
||
- `HandleResult`와 settlement는 one-terminal-call state machine으로 감싸고, DLQ는 publish confirmation 뒤
|
||
source ACK라는 기존 원칙을 강제한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- fake publisher 없이 full auto-config import → fake native client → actual `DefaultMessagePublisher` →
|
||
transport까지 1건 publish.
|
||
- pipeline 각 단계 실패가 pre-wire `REJECTED`와 post-wire `AMBIGUOUS`로 정확히 구분됨.
|
||
- admission/runtime lease가 success, async failure, timeout, cancellation 모두에서 exactly once 반환됨.
|
||
- handler success/retry/discard/throw, DLQ publish reject/ambiguous, settlement failure의 transition table 테스트.
|
||
|
||
### MSG-004 — Kafka consumer가 poll batch와 assignment epoch를 안전하게 관리하지 못한다
|
||
|
||
**근거**
|
||
|
||
- `KafkaConsumerRegistrar.pollOnce:154-175`는 poll이 반환한 전체 record를 순회하다 한 partition이 limit에
|
||
걸리거나 shutdown이 시작되면 현재 record를 seek하고 `break`한다. poll이 이미 반환한 다른
|
||
partition/뒤 record를 처리하거나 seek하지 않는다.
|
||
- `dispatch:183-205`의 `handlerPool.execute`가 `RejectedExecutionException`을 던지면 이미 획득한
|
||
partition permit/shutdown work count와 delivered offset을 정리하지 않는다.
|
||
- 같은 `dispatch:183-205`는 envelope decode와 `sink.apply(...).join()` 예외를 하나의 catch로 잡아
|
||
`PARKED` command를 넣는다. `applySettlements:210-213`는 실제 quarantine/DLQ write 없이 `PARKED`를
|
||
`COMPLETE`처럼 offset 완료 처리하므로 일시적 handler/DB 장애도 message loss가 된다.
|
||
- rebalance listener `114-122`는 handler가 끝나길 기다리지 않고 commit/forget한다.
|
||
- `KafkaSettlementCommand`에는 assignment generation/epoch가 없다. revoke 뒤 늦은 이전 handler의
|
||
settlement가 같은 partition의 새 assignment 상태에 적용될 수 있다.
|
||
- `QueuedSettlement:323-355`는 one-terminal CAS가 없고, poll thread가 실제 commit하기 전에
|
||
`SettlementResult.settled()`를 즉시 반환한다.
|
||
- `commitContiguous:235-254`는 `commitSync` 전에 `committed` map을 갱신하고 tracker를 prune한다.
|
||
commit 실패 후 다음 cycle이 해당 offset 재commit을 생략하거나 더 높은 watermark를 commit할 수 있다.
|
||
- public `pause/resume:272-285`와 `close:302-307`도 호출 thread에서 `Consumer`를 직접 만져 클래스가
|
||
선언한 poll-thread-only 불변식을 깬다.
|
||
- `ConsumerPolicy.handlerTimeout`은 선언돼 있지만 worker의 blocking `.join()`에 적용되지 않는다.
|
||
|
||
**실패 시나리오**
|
||
|
||
- partition 0의 첫 record가 limit에 걸린 순간 poll batch에 함께 온 partition 1 record를 잊는다. consumer
|
||
position은 이미 poll로 전진했으므로 rebalance/restart 전까지 처리 공백이 생길 수 있다.
|
||
- revoke된 epoch A의 handler가 늦게 ACK한 뒤 partition이 epoch B로 재할당되면 B에서 아직 처리하지 않은
|
||
offset이 complete/commit될 수 있다.
|
||
- `commitSync` 실패 전에 로컬 prune이 끝나면 이후 높은 offset commit이 실패 구간까지 포함해 message를
|
||
잃을 수 있다.
|
||
|
||
**구현 결정: partition state machine + assignment fencing**
|
||
|
||
1. `AssignmentEpoch(topicPartition, generation)`을 assignment마다 만들고 delivery/settlement command에
|
||
포함한다. current epoch가 아니면 stale settlement로 거절한다.
|
||
2. revoke 시 해당 partition을 pause하고 신규 dispatch를 막은 뒤 bounded deadline까지 in-flight를 drain,
|
||
성공한 contiguous prefix만 commit하고 state를 폐기한다. deadline 이후 delivery는 unsettled로 남긴다.
|
||
3. poll batch는 partition별로 처리한다. 제출하지 못한 **모든** record의 earliest offset을 partition별로
|
||
seek하고 해당 partition만 pause한다. 한 partition 때문에 다른 partition을 `break`하지 않는다.
|
||
4. executor rejection을 잡아 delivered 등록을 되돌리거나 아직 등록하기 전 submit을 시도하고,
|
||
coordinator/shutdown permit을 정확히 반환하며 해당 offset을 seek한다.
|
||
5. `QueuedSettlement`은 `AtomicReference<TerminalState>`로 acknowledge/requeue/discard 중 하나만 허용한다.
|
||
6. settlement future는 queue enqueue가 아니라 poll thread의 actual broker operation 결과로 완료한다.
|
||
7. `commitSync` 성공 뒤에만 local committed map과 tracker를 prune한다. 실패 시 tracker를 보존하고 retry
|
||
policy/health에 노출한다.
|
||
8. public pause/resume/close도 poll-thread command queue와 completion future로 직렬화한다.
|
||
9. decode failure는 confirmed quarantine/DLQ 뒤에만 source offset을 완료하고, handler exception은
|
||
`RetryDecisionEngine`으로 보낸다. handler deadline도 같은 typed outcome으로 처리한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 두 partition을 한 poll에 반환하고 한 partition만 limit에 걸리는 경우 다른 partition은 처리된다.
|
||
- executor rejection 후 in-flight=0, offset 미commit, 다음 poll 재전달.
|
||
- revoke → reassign → old handler ACK가 새 epoch watermark를 바꾸지 않음.
|
||
- ACK 두 번, ACK 후 requeue/discard는 두 번째 terminal call 거절.
|
||
- `commitSync` 첫 호출 실패/둘째 성공에서 local watermark가 성공 전 전진하지 않음.
|
||
- malformed delivery에서 DLQ unavailable이면 commit하지 않고, confirmed DLQ 뒤에만 commit.
|
||
- 다른 thread의 pause/resume/close가 consumer를 직접 호출하지 않고 poll queue에서 실행됨.
|
||
- never-completing handler가 configured timeout 뒤 policy대로 unsettled/retry됨.
|
||
- close/revoke deadline에서 unfinished handler의 offset이 commit되지 않음.
|
||
|
||
### MSG-005 — Rabbit publish/consume state machine이 native broker protocol을 완전히 표현하지 못한다
|
||
|
||
**publisher 근거**
|
||
|
||
- `RabbitMessagingTransport.publish:113-122`는 pending 등록 뒤 `channel.publish`를 호출한다. synchronous
|
||
예외를 잡지 않아 pending entry가 남고 메서드가 stage 대신 예외를 던진다.
|
||
- timeout은 `onConfirmTimeout:145-153` 외부 호출에 의존하지만 production scheduler/caller가 없다.
|
||
- `onConfirm:125-134`에는 Rabbit confirm의 `multiple` flag가 없다. 실제 IT listener도
|
||
`RabbitBrokerIT:110-112`에서 `multiple`을 버린다.
|
||
- `RabbitConfirmCoordinator:80-86`은 sequence 하나만 제거한다. multiple ACK/NACK는 `<= tag` 전체를
|
||
해결해야 한다.
|
||
- Rabbit return에는 publish sequence가 없다. IT가 `164-196`에서 `x-seq`를 수동 주입해 correlation하지만
|
||
`RabbitChannelPublisher` contract와 `RabbitPublishMapper`는 이를 보장하지 않고 production 구현도 없다.
|
||
- coordinator의 NACK는 `RabbitConfirmCoordinator:143-155`에서 `notTransmitted()`로 기록한다. broker가
|
||
frame을 받고 NACK한 결과이므로 transmission evidence와 모순된다.
|
||
- `close()`는 pending stage를 drain/complete하거나 native channel을 닫지 않는다.
|
||
|
||
**consumer 근거**
|
||
|
||
- `RabbitConsumerRegistrar.onMessage:103-126`의 하나의 catch가 decode, handler stage join, settlement
|
||
operation의 모든 `RuntimeException`을 잡아 `RABBIT_UNDECODABLE`로 discard한다.
|
||
- handler가 terminal settlement를 전혀 호출하지 않고 정상 complete해도 `true`를 반환한다.
|
||
- `close:156-159`는 `active=false`만 설정하지만 `onMessage:89-100`은 active flag를 검사하지 않아 close 뒤
|
||
delivery도 받아들일 수 있다.
|
||
- `RabbitSettlementController:74-82`는 native operation 호출 전에 settled flag를 세운다. operation이
|
||
전송 전에 synchronous failure하면 같은 delivery에 대한 안전한 재시도/상태 판단이 불가능하다.
|
||
|
||
**영향**
|
||
|
||
multiple ACK를 놓치면 pending publish가 영구 대기하고 memory가 증가한다. handler business failure나
|
||
ACK channel failure를 poison payload로 오인해 discard하면 재시도돼야 할 message를 잃는다.
|
||
|
||
**구현 결정: broker-owned publish/settlement state machine**
|
||
|
||
- adapter 내부에 실제 native `RabbitChannelBridge`를 구현해 sequence 예약, correlation header, mandatory
|
||
publish, confirm/return listener, scheduled timeout, channel-close drain을 한 객체가 소유한다.
|
||
- state event를 `Confirm(tag, multiple, ack)`, `Returned(correlation)`, `TimedOut`, `SendFailed`,
|
||
`ChannelClosed`로 모델링한다. ordered concurrent map에서 multiple이면 `headMap(tag, true)` 전체를
|
||
원자적으로 resolve한다.
|
||
- synchronous publish failure는 pending을 제거하고 typed classifier로 `REJECTED` 또는 보수적
|
||
`AMBIGUOUS` stage를 완료한다. native 예외를 caller에게 raw throw하지 않는다.
|
||
- NACK/return/timeout evidence 생성은 `RabbitPublishResultFactory` 하나로 통일한다.
|
||
- consumer는 decode try/catch만 좁게 잡는다. handler failure는 retry engine, settlement failure는
|
||
`UNKNOWN`/unsettled 경로로 보낸다. processor가 exactly-one terminal settlement를 강제한다.
|
||
- settlement는 `NEW -> IN_FLIGHT -> TERMINAL` state machine으로 두고, definite pre-send failure에서만
|
||
`NEW`로 복귀한다. close 뒤 `onMessage`는 즉시 false를 반환한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- multiple ACK/NACK가 tag 이하 pending 전체를 완료하고 pending=0.
|
||
- synchronous `basicPublish` throw, scheduled timeout, return-before-confirm, channel close 각각 pending=0.
|
||
- missing/malformed return correlation은 metric/audit와 conservative outcome으로 처리.
|
||
- handler throw/failed stage/settlement throw가 deserialization DLQ로 잘못 discard되지 않음.
|
||
- handler가 settlement 없이 complete하면 success로 ACK하지 않고 contract violation으로 처리.
|
||
- settlement synchronous failure와 close-after-delivery race에서 double ACK/discard가 없음.
|
||
|
||
### MSG-006 — outbox retry budget와 backoff가 relay 실행 경로에 연결되지 않는다
|
||
|
||
**근거**
|
||
|
||
- `OutboxProperties.java:21-29,66-73`은 `maxAttempts=10`을 설정한다.
|
||
- `OutboxRetryScheduler.java:53-101`은 pass backoff와 attempt exhaustion 판단을 구현한다.
|
||
- 그러나 `OutboxRelay.java:67-103`은 scheduler/maxAttempts를 받지 않고 모든 `AMBIGUOUS`를 즉시 다시
|
||
claim 가능하게 만든다. relay를 실제로 주기 실행하는 production lifecycle도 없다.
|
||
- claim SQL `JdbcOutboxRepository.java:48-69`에는 attempt cap이나 `next_attempt_at` 조건이 없다.
|
||
- `MessagingReliabilityAutoConfiguration.java:40-55`는 properties/scheduler bean을 만들지만 relay나
|
||
scheduling loop를 구성하지 않는다.
|
||
- `InboxCleanupJob`/`OutboxCleanupJob`은 batch size/max batches로 bounded라고 설명하지만 repository purge
|
||
port에 limit parameter가 없다. `JdbcInboxRepository.java:103-115`와
|
||
`JdbcOutboxRepository.java:206-218`은 cutoff 전체를 한 DELETE로 지워 큰 table에서 lock/WAL spike를
|
||
만들 수 있다.
|
||
|
||
**영향**
|
||
|
||
broker outage 동안 같은 backlog가 poll interval마다 반복 발행되어 broker와 DB를 더 압박한다. 문서와
|
||
settings는 10회 후 park를 약속하지만 실제 row는 `AMBIGUOUS`로 무한 재claim된다. relay 객체를 application이
|
||
직접 만들더라도 scheduler를 별도로 조립하지 않으면 같은 결과다.
|
||
|
||
**구현 결정**
|
||
|
||
1. retry clock은 process memory가 아니라 row의 `attempts`, `next_attempt_at`, last failure에 둔다.
|
||
2. claim predicate는 `next_attempt_at <= now`와 `attempts < maxAttempts`를 적용한다.
|
||
3. ambiguous transition에서 policy가 계산한 next-at을 함께 저장한다. attempt budget이 끝나면 별도
|
||
`PARKED`/`EXHAUSTED` terminal status를 사용한다. definite `REJECTED`와 attempt exhaustion을 같은
|
||
`FAILED`로 뭉개지 않는다.
|
||
4. pass-level outage backoff는 worker scheduler가 사용하되 row-level eligibility의 대체물이 아니다.
|
||
5. `SmartLifecycle` worker가 leader-only인지 all-replica `SKIP LOCKED`인지 명시한다. 후자를 택하면 cleanup
|
||
leader election은 별도 문제로 둔다.
|
||
6. purge port를 `purgeBefore(cutoff, limit)`로 바꾸고 PostgreSQL CTE에서 bounded ID를
|
||
`FOR UPDATE SKIP LOCKED`로 고른 뒤 delete한다. deleted count가 limit보다 작으면 sweep를 종료한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 연속 ambiguous에서 attempt/next-at이 증가하고 deadline 전 claim되지 않음.
|
||
- maxAttempts 도달 후 재claim되지 않으며 operator redrive만 가능.
|
||
- confirmed publish가 retry 상태를 지우고, definite rejected는 즉시 terminal.
|
||
- 두 relay가 outage 중에도 같은 row를 동시에 publish하지 않고 설정된 rate bound를 넘지 않음.
|
||
- retention row가 batch보다 많아도 한 SQL call의 delete count가 limit 이하이고 concurrent append/claim을
|
||
장시간 block하지 않음.
|
||
|
||
### MSG-007 — 하나의 starter가 두 broker와 모든 선택 기능을 노출하면서도 자기완결적이지 않다
|
||
|
||
**근거**
|
||
|
||
- `messaging-spring-boot-starter/build.gradle:4-19`는 Kafka, Rabbit, 두 JDBC reliability, admin 등 16개
|
||
internal leaf를 전부 `api`로 노출한다.
|
||
- AutoConfiguration imports는 core/Kafka/Rabbit/reliability/admin 다섯 구성을 항상 후보로 등록한다.
|
||
- starter가 두 broker client를 끌어오므로 양쪽 `@ConditionalOnClass`가 동시에 참이 된다.
|
||
- Kafka/Rabbit security configurer는 `CredentialRuntimeRegistry`를 필수 parameter로 받지만
|
||
(`KafkaMessagingAutoConfiguration.java:70-74`, `RabbitMessagingAutoConfiguration.java:56-60`), registry는
|
||
`CredentialProvider`가 있을 때만 생긴다 (`MessagingCoreAutoConfiguration.java:216-220`).
|
||
- 현재 `MessagingAutoConfigurationTest.java:41-45,119-131`은 core config만 로드하고 fake publisher를
|
||
항상 제공해 전체 imports/conditional graph를 검증하지 않는다.
|
||
|
||
**구현 결정: 기능별 starter + broker runtime Abstract Factory**
|
||
|
||
```text
|
||
messaging-spring-boot-autoconfigure-core
|
||
messaging-spring-boot-starter-core
|
||
messaging-spring-boot-starter-kafka
|
||
messaging-spring-boot-starter-rabbit
|
||
messaging-spring-boot-starter-reliability-jdbc-postgresql
|
||
messaging-spring-boot-starter-admin
|
||
```
|
||
|
||
- core starter는 broker SDK, JDBC, admin을 전이 의존하지 않는다.
|
||
- Kafka/Rabbit starter가 자기 native client factory, profile binder/validator, runtime factory를 소유한다.
|
||
- 일반 internal dependency는 `implementation`으로 낮추고 실제 public signature에 나타나는 타입만 `api`로
|
||
노출한다.
|
||
- credential이 필수인 production profile은 명확한 startup error로 실패한다. local/insecure profile을
|
||
지원한다면 별도 opt-in이고 production flag와 동시에 허용하지 않는다.
|
||
- 기존 통합 artifact 이름을 유지해야 하면 all-in-one runtime이 아니라 BOM/dependency constraints로
|
||
남긴다.
|
||
|
||
**필수 테스트**
|
||
|
||
- published imports 전체를 fake publisher/credential 없이 로드했을 때 의도한 error 한 개로 실패.
|
||
- core-only, Kafka-only, Rabbit-only classpath fixture compile/context.
|
||
- Kafka-only에서 Rabbit config/type가 없고 반대도 동일.
|
||
- 두 broker profile이 있을 때 destination별 router가 명시적으로 선택하며 duplicate broker ID는 실패.
|
||
- optional admin/reliability가 dependency를 추가하지 않으면 bean도 endpoint도 생기지 않음.
|
||
|
||
### MSG-008 — 설정 namespace 세 개와 dead flags 때문에 문서대로 구성할 수 없다
|
||
|
||
**근거**
|
||
|
||
| 위치 | namespace | 실제 상태 |
|
||
|---|---|---|
|
||
| 기존 runtime `MessagingSettings.java:14`, `application.yml:733-740` | `app.messaging` | 현재 배포/환경변수 정본 |
|
||
| 신규 `MessagingProperties.java:14` | `backend.messaging` | experimental/bridge/backpressure/shutdown만 binding |
|
||
| `docs/messaging/configuration-reference.md:6,80,99,120,138,150` | `messaging` | destination/broker/security 포함하지만 코드 binder 없음 |
|
||
|
||
- 신규 properties에는 destination, broker, security profile이 없다 (`MessagingProperties.java:17-20`).
|
||
- experimental/bridge getter는 production source에서 읽히지 않고 테스트만 기본값을 확인한다. 관련
|
||
experimental/bridge leaf도 starter dependency가 아니므로 flag를 true로 해도 adapter가 활성화되지 않는다.
|
||
- `ValidatedDestinationRegistry`는 bound settings가 아니라 application이 이미 bean으로 제공한
|
||
`DestinationProfile`만 수집한다 (`MessagingCoreAutoConfiguration.java:56-63`).
|
||
- Kafka/Rabbit/security validator는 bean으로 존재할 뿐 registry startup validation에 참여하지 않는다.
|
||
- `DestinationProfileValidator.validateAll:130-133`은 retry graph와 DLQ graph를 따로 순회해
|
||
`A.retry -> B`, `B.dlq -> A` 같은 mixed-edge cycle을 놓칠 수 있다.
|
||
|
||
**구현 결정: 하나의 typed configuration compiler**
|
||
|
||
- 이미 배포 계약과 `APP_MESSAGING_*`가 존재하므로 이번 cutover의 canonical prefix는
|
||
**`app.messaging`**로 유지한다. 장기적으로 `ca-skeleton.messaging`로 바꾸고 싶다면 별도 ADR/한정된
|
||
migration release로 다루며 지금 세 번째 namespace를 추가하지 않는다.
|
||
- mutable nested bean보다 immutable validated settings를 사용하고 destination/broker/security map을
|
||
모두 표현한다. unknown field는 거절한다.
|
||
- binder 결과를 `MessagingConfigurationCompiler`가 immutable `DestinationProfile`과 broker runtime
|
||
descriptor로 컴파일한다.
|
||
- generic/Kafka/Rabbit/security validator는 `ProfileValidator` Strategy 목록을 받는 Composite로 묶어
|
||
startup에서 모든 profile에 적용한다.
|
||
- retry/DLQ edge label을 가진 하나의 directed graph를 만들고 단일 DFS/SCC cycle validation을 수행한다.
|
||
admin topology validator도 같은 compiled graph/report contract를 사용한다.
|
||
- `backend.messaging` 또는 bare `messaging` key가 발견되면 actionable migration error로 실패한다.
|
||
한 release alias를 허용해도 두 prefix 동시 사용은 거절한다.
|
||
- experimental/bridge flag는 실제 module/bean activation에 연결하거나 공개 설정에서 제거한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 문서 YAML 전체를 fixture로 읽어 destination/broker/security/runtime descriptor가 정확히 생성됨.
|
||
- typo/unknown field, dangling DLQ, missing broker, unsafe TLS/auth, broker capability mismatch가 boot failure.
|
||
- retry와 DLQ edge를 섞은 cycle이 startup에서 정확한 경로와 함께 거절됨.
|
||
- old/new prefix 단독·혼합 case와 configuration metadata snapshot.
|
||
- experimental flag false/true에서 실제 bean graph가 각각 없고/있으며 dependency 부재 시 명시적 실패.
|
||
|
||
### MSG-009 — batch timeout과 stop-on-first-rejection의 공개 계약이 구현되지 않는다
|
||
|
||
**근거**
|
||
|
||
- `BatchPublishOptions.java:14-19`는 timeout을 whole-batch deadline으로 정의한다.
|
||
- `DefaultBatchMessagePublisher.java:53-90`은 `options.timeout()`을 읽지 않고 `allOf`를 무기한 기다린다.
|
||
- `70-81`은 빠른 for-loop에서 각 future가 그 순간 이미 done일 때만 rejection을 본다. 일반적인 비동기
|
||
broker rejection이 도착하기 전에 나머지 요청을 모두 제출한다.
|
||
- 현재 async failure test도 이미 완료된 failed future를 사용하고 진짜 delayed rejection 뒤 미제출을
|
||
검증하지 않는다.
|
||
|
||
**구현 결정**
|
||
|
||
먼저 option 의미를 두 mode로 명확히 한다.
|
||
|
||
1. `BEST_EFFORT_CONCURRENT`: bounded concurrency로 전부 제출하고 whole-batch deadline에서 미확정 항목을
|
||
`AMBIGUOUS/TIMEOUT`으로 완료한다. stop-on-first를 허용하지 않는다.
|
||
2. `STOP_AFTER_FIRST_REJECTION`: 순차 또는 작은 bounded window로 제출하고 최초 terminal rejection 뒤 아직
|
||
시작하지 않은 항목은 `NOT_SUBMITTED_AFTER_REJECTION`으로 결과에 포함한다. 이미 in-flight인 항목은
|
||
취소했다고 성공/실패를 추측하지 않고 deadline까지 evidence를 기다린다.
|
||
|
||
`BatchPublishResult`는 입력 index마다 정확히 한 item result를 가져야 한다. `break`로 결과 개수를 줄이면
|
||
caller가 미제출과 결과 유실을 구분할 수 없다. Java 21 `StructuredTaskScope`를 API에 새로 노출할 필요는
|
||
없으며 bounded executor/semaphore와 deadline scheduler로 충분하다.
|
||
|
||
**필수 테스트**
|
||
|
||
- never-completing publisher가 batch timeout에 전체 stage를 완료하고 outstanding을 ambiguous로 표시.
|
||
- delayed first rejection 후 아직 시작하지 않은 index는 호출되지 않고 explicit not-submitted result.
|
||
- 이미 in-flight success/reject/ambiguous는 결과에 보존.
|
||
- empty batch, max size 경계, executor rejection, caller cancellation에서 permit/thread leak 없음.
|
||
|
||
### MSG-010 — runtime lifecycle과 backpressure가 boolean counter API에 의존해 손상될 수 있다
|
||
|
||
**근거**
|
||
|
||
- `BackpressureController.release:73-80`은 destination entry가 없거나 이미 0이어도 global이 양수면 global을
|
||
감소시킨다. 잘못된 destination/double release로 global limit를 우회할 수 있고 0 counter entry도
|
||
map에 남는다.
|
||
- `GracefulShutdownCoordinator.endWork:62-65`도 double release에서 음수가 될 수 있다.
|
||
- `DefaultMessagingRuntimeRegistry.closeExpiredDraining:83-110`은 모든 retired generation에 caller가 준
|
||
하나의 `retiredAt`을 적용한다. `Generation:124-130`에는 자체 retirement time이 없다.
|
||
- idle close된 generation도 draining list에서 즉시 제거되지 않는다. `closeExpiredDraining`과
|
||
`beginDrain`의 production caller/`SmartLifecycle` 연결이 없다.
|
||
|
||
**구현 결정: RAII-style Permit + generation-owned lifecycle**
|
||
|
||
```java
|
||
interface AdmissionPermit extends AutoCloseable {
|
||
DestinationName destination();
|
||
@Override void close(); // CAS, exactly once
|
||
}
|
||
|
||
interface WorkPermit extends AutoCloseable {}
|
||
```
|
||
|
||
- acquire가 boolean 대신 destination에 묶인 permit을 반환하고 caller는 `try/finally` 또는 async
|
||
`whenComplete`에서 close한다. 임의 destination 문자열 release API는 제거한다.
|
||
- per-destination counter가 0이면 conditional remove한다. global/per-destination 변경이 항상 한 permit의
|
||
lifecycle로 짝을 이뤄야 한다.
|
||
- generation은 `retiredAt`, monotonic generation ID, active leases, closed flag를 보유한다. `Clock`을
|
||
registry에 주입하고 `closeExpiredDraining(now)`가 generation별 deadline을 판단한다.
|
||
- registry는 `AutoCloseable`/lifecycle을 구현해 current와 draining runtime 모두 exactly once close한다.
|
||
- Spring lifecycle 순서는 consumer registration 중단 → admission close → begin drain → deadline wait →
|
||
unresolved work를 unsettled/ambiguous로 종료 → transport close다.
|
||
|
||
**필수 테스트**
|
||
|
||
- wrong destination/double close/concurrent close가 count를 손상하지 않음.
|
||
- 서로 다른 시각에 retired된 두 generation이 자기 deadline에만 close.
|
||
- 마지막 lease 반환 시 draining 목록에서 즉시 제거, leaked lease는 deadline에 force-close.
|
||
- context close 중 신규 publish/delivery 거절, 기존 work drain, close exactly once.
|
||
|
||
### MSG-011 — credential rotation과 broker security configuration이 fail-closed하지 않다
|
||
|
||
**근거**
|
||
|
||
- `CredentialRuntimeRegistry.resolve:58-69`는 `get` → 외부 fetch → `put` → 이전 runtime `clear`를
|
||
synchronization 없이 수행한다.
|
||
- 두 caller가 동시에 rotation하면 둘 다 같은 old runtime을 보고 replacement를 fetch한 뒤 하나가 map에서
|
||
유실되고 clear되지 않을 수 있다.
|
||
- 조금 늦은 caller는 첫 replacement를 current로 잡고 두 번째 replacement를 넣은 뒤, 첫 caller가 아직
|
||
쓰는 credential material을 clear할 수 있다.
|
||
- `CredentialRuntime.material/clear:81-85,128-132` 자체도 동기화되지 않는다.
|
||
- `BrokerTlsPolicy.java:72-96`은 알려진 구버전 denylist 방식이라 `SSL`, `TLSv0.9`, `PLAINTEXT` 같은
|
||
비정상 protocol 문자열도 통과할 수 있다.
|
||
- `KafkaSecurityConfigurer.java:124-133`은 password를 quoted JAAS string에 escaping 없이 삽입한다.
|
||
quote/backslash/semicolon/newline이 있는 정상 secret도 parsing을 깨거나 option injection이 된다.
|
||
- OAuth와 mTLS branch는 credential을 resolve하거나 mechanism 이름만 설정하고 실제 callback/client
|
||
identity 구성까지 완결하지 않는다. 구현되지 않은 mode를 부분 설정으로 허용하면 연결 시점에 실패한다.
|
||
|
||
**구현 결정**
|
||
|
||
- credential을 raw cached value가 아니라 `CredentialGenerationLease`로 대여한다.
|
||
- key별 single-flight refresh(`ConcurrentHashMap.compute` 또는 keyed lock)로 한 replacement만 publish한다.
|
||
- 이전 generation은 새 client/runtime가 성공적으로 설치된 뒤 draining으로 보내고 active lease가 0이거나
|
||
deadline이 지났을 때만 clear한다. 이는 messaging runtime generation과 같은 lifecycle에 결합한다.
|
||
- provider fetch 실패 시 아직 만료되지 않은 current credential을 정책에 따라 유지하고, 이미 만료된
|
||
credential은 fail-closed한다. 이 정책을 typed outcome으로 기록한다.
|
||
- material clone의 소유권과 clear 책임을 문서화하고 char array가 map에서 유실되지 않도록 한다.
|
||
- TLS protocol은 `TLSv1.2`/`TLSv1.3` allowlist로 검증하고 hostname verification/production auth를 함께
|
||
compiler invariant로 둔다.
|
||
- JAAS 문자열 직접 조립 대신 표준 escaping 또는 callback handler/typed client property를 사용한다.
|
||
OAuth/mTLS는 실제 client context까지 구성하기 전 profile validation에서 명시적으로 거절한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 100 concurrent resolve에서 provider refresh 1회, winning generation 1개, losing secret 0개.
|
||
- old lease 사용 중 rotation해도 material 접근 가능; lease close 뒤 zeroization.
|
||
- fetch failure before/after expiry, shutdown clear, double lease close.
|
||
- quote/backslash/semicolon/newline secret, protocol case/unknown/SSL/TLSv0.9, incomplete OAuth/mTLS profile.
|
||
|
||
### MSG-012 — wire-facing value object/header 경계가 control character와 크기 공격을 허용한다
|
||
|
||
**근거**
|
||
|
||
- `MessageId.java:13-19`는 UUIDv7이라고 문서화하지만 constructor는 모든 UUID version/variant를 받는다.
|
||
- `MessageType`, `ProducerId`, `CorrelationId`는 blank와 Java character count만 검사한다. control character,
|
||
newline, unbounded UTF-8 byte expansion을 허용한다.
|
||
- `ContentType.java:26-34`는 slash 포함 여부와 lowercase만 검사해 유효한 media type 문법을 보장하지 않는다.
|
||
- `TraceContext.java:17-24`는 Optional non-null 외 W3C grammar/size 제한이 없다.
|
||
- `MessageEnvelope.java:42-43,49-65`의 partition/ordering key는 길이 제한이 없다.
|
||
- `HeaderName.java:18-35`는 nonblank/128 UTF-8 bytes만 검사해 CRLF, NUL, colon, leading/trailing
|
||
whitespace를 허용한다. secret/reserved denylist는 lowercase exact match라 `Authorization ` 같은 변형이
|
||
우회한다.
|
||
- `JdbcOutboxRepository.toJson/fromJson:262-324`는 header JSON을 손으로 처리하며 quote/backslash 외 JSON
|
||
control escape를 하지 않는다. 허용된 newline/NUL은 PostgreSQL JSONB insert 실패 또는 lossy parsing을
|
||
만들 수 있다.
|
||
- Kafka/Rabbit header mapper는 검증된 것으로 가정하고 이 이름/값을 broker wire에 전달한다.
|
||
|
||
**구현 결정: canonical wire-safe value objects**
|
||
|
||
- identifier/header name은 ASCII token grammar와 UTF-8 byte limit를 사용한다. name은 trim을 허용하지 않고
|
||
canonical lowercase를 저장한다. 허용 문자를 정본 regex로 하나만 둔다.
|
||
- reserved/secret 검사는 canonical name과 prefix/구조 규칙으로 수행한다. metric-safe라는 문구가 실제
|
||
low-cardinality를 의미하지는 않으므로 producer/message type의 registry allowlist도 별도 둔다.
|
||
- header value는 CR/LF/NUL 등 transport-unsafe controls를 거절하고 broker별 encoded byte budget을
|
||
envelope 전체 size budget에 포함한다.
|
||
- W3C Trace Context parser로 traceparent/tracestate/baggage grammar와 표준 size/member bound를 적용한다.
|
||
- partition/ordering key는 typed `PartitionKey`/`OrderingKey`로 만들고 bytes bound를 둔다.
|
||
- UUIDv7만 받으려면 version/variant를 검증한다. 임의 UUID import가 필요하면 문서와 타입명을 일반
|
||
`MessageId`로 정직하게 바꾸고 `newId()`만 v7임을 명시한다.
|
||
- outbox JSON은 검증된 JSON serializer/PG JSONB mapping을 사용한다. codec dependency 회피를 위해
|
||
correctness를 포기하지 않는다.
|
||
|
||
**필수 테스트**
|
||
|
||
- CRLF/NUL/colon/space/Unicode confusable/UTF-8 max+1 header와 identifier.
|
||
- `Authorization `, mixed whitespace/control, reserved prefix 변형이 모두 거절됨.
|
||
- valid/invalid W3C trace vectors와 total baggage bound.
|
||
- Kafka/Rabbit/outbox round-trip에서 canonical header가 byte-for-byte 보존됨.
|
||
|
||
### MSG-013 — diagnostics 값이 cardinality guard 뒤에서 metric tag로 추가된다
|
||
|
||
**근거**
|
||
|
||
- `MessagingMetrics.recordDiagnostics:115-132`는 base `MessagingTags`만 `admitted(tags)`로 guard한 뒤,
|
||
arbitrary diagnostic key/value를 `diagnostic`/`value` tag로 추가한다.
|
||
- redaction은 알려진 secret key를 가릴 뿐 고유 message ID, exception message, URL, tenant 값의 무한
|
||
cardinality를 막지 않는다.
|
||
- `CardinalityGuard.admit:50-61`의 size-check/add는 원자적이지 않고, `admit(MessagingTags):70-76`은 뒤
|
||
dimension에서 실패해도 앞 dimension을 이미 관측 집합에 추가한다.
|
||
- 실제 central pipeline에 `MessagingMetrics`가 조립되지 않아 일부 contract는 사용되지 않는다.
|
||
|
||
**영향**
|
||
|
||
한 request마다 다른 diagnostic value가 meter series를 영구 생성해 metric backend와 application heap을
|
||
소진할 수 있다. 민감 값이 알려진 key가 아닌 자유 text에 포함되면 redactor도 막지 못한다.
|
||
|
||
**구현 결정**
|
||
|
||
- metric tag는 destination/broker/operation/outcome/failure-code처럼 닫힌 allowlist와 bounded vocabulary만
|
||
허용한다.
|
||
- arbitrary diagnostics는 sanitized structured log/trace event로 보내고 metric에는 fixed counter와
|
||
normalized failure code만 남긴다.
|
||
- guard는 안전망이지 동적 사용자 값을 허용하는 근거가 아니다. 필요하면 dimension별 synchronized
|
||
bounded set/atomic compute를 사용하고 tag set 전체를 preflight한 뒤 commit한다.
|
||
- `MessagingObservation`을 중앙 publisher/delivery processor decorator에 조립한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 10,000개 고유 diagnostic value를 보내도 meter count가 고정.
|
||
- secret이 key/value/free-form exception 어느 위치에서도 tag/log에 평문 노출되지 않음.
|
||
- concurrent cardinality admission이 configured limit를 넘지 않고 rejected counter가 정확함.
|
||
|
||
### MSG-014 — Stable과 live-broker coverage가 실행 결과가 아니라 hard-coded 자기선언이다
|
||
|
||
**근거**
|
||
|
||
- `BrokerFailureMatrix.shipped:97-129`는 5개 scenario 모두를 Kafka/Rabbit `LIVE_BROKER`로 하드코딩한다.
|
||
- `CrossBrokerContractSuite.java:31-53`은 실제 JUnit/CI artifact가 아니라 그 map을 assert한다.
|
||
- `RabbitBrokerIT.java:129-162`의 live test는 routable/unroutable happy contract 네 개뿐이며
|
||
connection-refused, cut-after-write, confirm-timeout, settlement-lost, high-latency 5개를 실행하지 않는다.
|
||
- Docker가 없으면 `DockerAvailability` + `@EnabledIf`로 IT 전체가 skip되지만, messaging 전용 fail-closed
|
||
workflow/evidence gate가 없다.
|
||
- `CompatibilityMatrix.java:62-67`와 `docs/messaging/support-matrix.md:8-13`은 Kafka 4.2/4.3을 인증했다고
|
||
선언한다. 실제 `KafkaContainerFixture.java:38`, `KafkaBrokerIT.java:74`,
|
||
`KafkaAmbiguityChaosIT.java:59`는 `apache/kafka:4.1.0`, lockfile client는 4.1.1이다.
|
||
|
||
**영향**
|
||
|
||
Docker가 전혀 없는 CI와 선언 버전을 한 번도 실행하지 않은 build도 Stable gate를 통과한다. matrix와
|
||
문서가 서로 일치하는 테스트는 둘이 같은 잘못된 상수를 복제했는지만 증명한다.
|
||
|
||
**구현 결정: evidence manifest release gate**
|
||
|
||
1. dev `test`는 Docker 부재 시 skip할 수 있지만 별도 `messagingCertificationTest`/workflow는 Docker와
|
||
각 broker version이 없으면 실패한다.
|
||
2. broker/version/scenario/test commit/image digest/result/timestamp를 machine-readable JSON artifact로
|
||
생성한다.
|
||
3. compatibility/failure matrix는 source 상수가 아니라 해당 release run의 signed/immutable evidence를
|
||
소비한다. 증거가 없으면 `NOT_COVERED`다.
|
||
4. Kafka 4.2와 4.3 이미지가 실제 존재하고 프로젝트가 지원할 준비가 됐을 때 각각 matrix job으로 실행한다.
|
||
그 전에는 현재 검증한 4.1.x만 표기하거나 Stable 주장을 내린다.
|
||
5. Rabbit도 5개 network fault와 consumer settlement loss를 실제 broker/proxy에서 실행한다.
|
||
6. `failOnNoDiscoveredTests`, expected suite count, skip count zero를 release lane에서 강제한다.
|
||
|
||
**필수 artifact**
|
||
|
||
```json
|
||
{
|
||
"adapter": "messaging-rabbit",
|
||
"brokerVersion": "4.3.x",
|
||
"scenario": "confirm-timeout",
|
||
"testId": "...",
|
||
"outcome": "AMBIGUOUS",
|
||
"imageDigest": "sha256:...",
|
||
"gitCommit": "..."
|
||
}
|
||
```
|
||
|
||
현재 로컬 run에서 Docker IT 10개가 실제 실행되어 skip 0이었던 사실은 긍정적이지만, 위 누락 scenario와
|
||
version을 대신하지 않는다.
|
||
|
||
### MSG-015 — 기존 application/runtime과 신규 platform 사이에 semantic bridge와 cutover authority가 없다
|
||
|
||
**근거**
|
||
|
||
- 현재 `app-bootstrap` dependency/runtime membership은 기존 `adapter:outbound:messaging`만 포함한다.
|
||
신규 24개 leaf는 모두 `runtime_memberships: []`이고 `dev.caskeleton.messaging.*`를 소비하는 production
|
||
bridge가 신규 tree 밖에 없다.
|
||
- application canonical port는 `OutboxMessagePublishPort.java:10-19`, 기존 adapter API는 별도
|
||
`core/MessagePublisher`, 신규 API는 `messaging-core-api/.../publish/MessagePublisher.java:13-25`, reliability
|
||
API에는 다시 `ReliableMessagePublisher`가 있다.
|
||
- 기존 `ValidatedIntegrationEvent`/`OutboxEvent`/v2 모델과 신규 `MessageEnvelope`/`OutboxRecord`가
|
||
identity, metadata, wire bytes를 서로 다르게 표현한다.
|
||
- 기존 `OutboxEventStatus.java:25-36`의 `FAILED`는 retryable이고 `DEAD`가 terminal이다. 신규
|
||
`OutboxStatus`의 `AMBIGUOUS`는 retryable이며 `FAILED`는 definite rejection terminal이다.
|
||
- application-core가 신규 platform implementation/API를 직접 import하면 현재 registry와 local layer
|
||
policy를 위반한다.
|
||
|
||
**영향**
|
||
|
||
이름이 같은 enum을 기계적으로 매핑하면 retryable/terminal 의미가 뒤집힌다. 두 publisher/outbox writer를
|
||
동시에 켜면 한 business fact가 두 durable store와 두 relay로 발행된다. 신규 adapter의 AMBIGUOUS를 기존
|
||
exception 하나로 축약하면 broker에 저장됐을 수 있는 message의 정합성을 잃는다.
|
||
|
||
**구현 결정: Anti-Corruption Layer + single publication authority**
|
||
|
||
1. application-owned semantic port/model을 canonical business boundary로 유지한다.
|
||
2. application outcome에 최소 `CONFIRMED`, `AMBIGUOUS`, `REJECTED_BEFORE_SEND`,
|
||
`REJECTED_AFTER_BROKER`를 표현하고 기존 state transition 표를 먼저 확정한다.
|
||
3. outbound `messaging-platform-bridge`가 validated application event를 canonical platform envelope로
|
||
변환하고 신규 `PublishResult`를 application outcome으로 역변환한다. application은 신규 타입을 모른다.
|
||
4. event/message ID, type, schema revision, partition/order/correlation/causation/tenant/trace, exact payload
|
||
digest와 wire version을 golden contract로 보존한다.
|
||
5. 기존 `OutboxPublicationAuthority`/dispatch fence를 재사용해 writer와 relay authority는 항상 하나만
|
||
ACTIVE가 되게 한다. dual write/publish는 금지한다.
|
||
6. 첫 cutover는 기존 outbox storage/writer를 유지하고 **transport만** 신규 platform으로 바꾼다.
|
||
storage migration은 별도 release에서 shadow read → authority switch → old backlog drain 순서로 한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- old event → bridge → new envelope golden bytes/headers/digest.
|
||
- 모든 new publish outcome × old outbox state transition table.
|
||
- 한 business action당 durable row와 broker publish가 정확히 하나.
|
||
- mixed-version rolling deployment에서 authority switch, crash, rollback.
|
||
- ArchUnit로 deprecated model에 신규 production import 금지.
|
||
|
||
### MSG-016 — broker round-trip과 outbox/CDC가 canonical envelope 정보를 유실하거나 reserved 값을 위조한다
|
||
|
||
**근거**
|
||
|
||
- Kafka/Rabbit producer header mapper는 identity/trace 등을 쓰고 reverse mapper도 제공하지만,
|
||
`KafkaDeliveryMapper.java:72-92`와 `RabbitDeliveryMapper.java:73-96`는 tenant를 empty, application headers를
|
||
empty, encoded schema reference를 empty로 재구성한다.
|
||
- `KafkaRetryMetadataMapper`는 envelope retry header를 읽는데 Kafka delivery mapper가 headers를 비우므로
|
||
retry attempt가 다시 1로 시작할 수 있다. Rabbit의 malformed retry header도 `attemptOf:132-141`에서
|
||
fail-closed하지 않고 1로 되돌린다.
|
||
- `OutboxRecord`는 arbitrary `Map<String,String>`을 받고 `OutboxEnvelopeFactory`는 이를
|
||
`MessageHeaders.platform`으로 만든다. 이 경로는 reserved header를 허용한다.
|
||
- Kafka mapper는 canonical reserved headers를 먼저 추가한 뒤 `KafkaHeaderMapper.java:70`에서 envelope
|
||
headers를 다시 추가하고 consumer는 `lastHeader:100-104`를 신뢰한다. DB의 forged `msg.id`가 canonical
|
||
message ID를 덮을 수 있다. Rabbit도 `RabbitHeaderMapper.java:39-87`에서 application header를 마지막에
|
||
쓴다.
|
||
- 신규 `OutboxRecord`에는 producer/correlation/causation/tenant/trace/schema reference가 없고
|
||
`OutboxEnvelopeFactory`가 일부 값을 주입하거나 empty로 발명한다.
|
||
- `DebeziumOutboxEventRouter`는 destination을 event key로 사용하며, polling mapper와 test-only CDC mapper가
|
||
partition key/header 규칙을 다르게 적용한다.
|
||
|
||
**영향**
|
||
|
||
tenant isolation, inbox identity, retry cap, ordering, tracing, schema resolution이 publish 방식
|
||
(direct/polling/CDC)과 broker에 따라 달라진다. forged reserved ID는 다른 message의 inbox dedup/audit을
|
||
오염시킬 수 있다.
|
||
|
||
**구현 결정: canonical EnvelopeHeaderCodec + persistence schema**
|
||
|
||
- `ApplicationHeaders`와 `PlatformHeaders`를 타입 수준에서 분리한다. application code/outbox input은
|
||
reserved namespace를 생성할 수 없어야 한다.
|
||
- `EnvelopeHeaderCodec` 하나가 canonical reserved field와 application header를 encode/decode한다.
|
||
Kafka, Rabbit, outbox polling, Debezium SMT, CloudEvents가 같은 contract를 사용한다.
|
||
- duplicate reserved header, malformed retry/tenant/schema/trace 값은 reject/quarantine한다. first/last wins로
|
||
복구하지 않는다.
|
||
- outbox에 canonical metadata column 또는 versioned canonical envelope bytes를 저장한다. CDC event key는
|
||
partition key(없으면 message ID 등 명시된 fallback), destination은 routing metadata로 분리한다.
|
||
- `ReservedHeaders`에 tenant/schema reference/retry/redrive fields를 명시하고 typed decode 결과를 사용한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- full envelope의 Kafka/Rabbit property-based round-trip: 모든 metadata/application headers/payload 동일.
|
||
- retry 1 → 2 → 3 → DLQ, malformed retry header quarantine.
|
||
- forged/duplicate `msg.id`, type, correlation, tenant header가 broker/outbox 경계에서 거절됨.
|
||
- 실제 PostgreSQL + Debezium container에서 polling과 CDC consumer가 key/header/payload byte-for-byte 동일.
|
||
- tenant A message가 tenant B context로 재구성되지 않음.
|
||
|
||
### MSG-017 — PublishOptions, capability와 PublishResult가 실제 adapter 동작보다 강한 계약을 노출한다
|
||
|
||
**근거**
|
||
|
||
- `PublishOptions.java:16-25`는 timeout, confirmation, deduplication, broker hints를 정의하지만
|
||
`KafkaMessagingTransport.publish:106-137`와 `RabbitMessagingTransport.publish:95-122`는
|
||
`request.options()`를 사용하지 않는다.
|
||
- `PublishOptions.isM1Compatible()`은 production에서 호출되지 않아 일반 caller가 M3 broker hints를 넣을 수
|
||
있다.
|
||
- Kafka capability는 deduplicated publish를 true로 선언하지만 producer idempotence는 동일 producer
|
||
session의 sequence retry를 다루며, message ID 기반 process restart 간 dedup을 보장하지 않는다.
|
||
- `PublishResult` public constructor는 completion/evidence/routing의 모순 조합을 만들 수 있다. exception
|
||
conversion도 typed transmission evidence가 없는 일부 exception을 `NOT_TRANSMITTED`로 단정한다.
|
||
|
||
**구현 결정: normalized options + runtime-derived capabilities + sealed outcome**
|
||
|
||
- 중앙 publisher가 destination policy와 call option을 merge하는 `ResolvedPublishOptions`를 만든다.
|
||
unsupported confirmation/dedup/hint는 broker에 보내기 전에 명시적으로 거절한다.
|
||
- timeout은 admission부터 broker outcome까지 absolute deadline으로 전파하고 timeout 시 transmission
|
||
milestone에 따라 AMBIGUOUS를 판단한다.
|
||
- capability는 adapter 상수가 아니라 destination + 실제 producer/runtime config에서 계산한다. persistent
|
||
message-ID store가 없다면 Kafka `deduplicatedPublish=false`다.
|
||
- public arbitrary constructor 대신 private factories/sealed ADT를 사용한다.
|
||
|
||
```text
|
||
Confirmed
|
||
RejectedBeforeTransmission
|
||
RejectedAfterBrokerAcceptance
|
||
Ambiguous
|
||
```
|
||
|
||
- broker hint map은 M3 native API의 typed option으로 이동하거나 일반 M1 API에서 제거한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- per-call timeout/confirmation/dedup/hint 지원·미지원 matrix.
|
||
- producer restart 뒤 같은 message ID가 broker 중복 억제되지 않음을 capability test로 고정.
|
||
- 모든 result truth table과 invalid combination compile/construction 불가.
|
||
- synchronous native send exception과 callback exception이 evidence-bearing typed outcome으로 변환됨.
|
||
|
||
### MSG-018 — Kafka transactional processor가 handler를 transaction 시작 전에 실행한다
|
||
|
||
**근거**
|
||
|
||
- `KafkaTransactionalProcessor.java:20-28`은 handler callback이 Kafka transaction 안에서 실행된다고
|
||
설명한다.
|
||
- `SpringKafkaTransactionalProcessor.java:47-56`은 handler를 먼저 호출한 뒤 publisher로 넘긴다.
|
||
- 실제 `producer.beginTransaction()`은 `KafkaTransactionalPublisher.java:94`에서 나중에 실행된다.
|
||
|
||
**영향**
|
||
|
||
handler가 직접 만든 Kafka output이나 외부 side effect는 문서와 달리 transaction 밖이다. handler 성공 뒤
|
||
begin/commit 실패, handler 도중 예외, dynamic output 생성에서 input offset과 output visibility가 한
|
||
transaction이라는 보장이 성립하지 않는다.
|
||
|
||
**구현 결정**
|
||
|
||
- transaction owner가 `beginTransaction`한 뒤 callback을 실행한다.
|
||
- callback은 raw producer를 받지 않고 `ProcessResult(value, outputs)` 또는 transaction-scoped output port를
|
||
사용한다.
|
||
- callback output 전송 + source offsets + commit을 한 try 안에 두고 모든 failure에서 abort한다.
|
||
- 외부 DB side effect와 Kafka transaction을 exactly-once로 묶는다고 표현하지 않는다. DB 효과는 inbox/
|
||
outbox 등 별도 idempotency가 필요하다.
|
||
|
||
**필수 테스트**
|
||
|
||
- handler 안에서 producer transaction state가 active.
|
||
- handler throw, output send failure, offset send failure, commit failure마다 abort되고 `read_committed`에서
|
||
output/offset이 보이지 않음.
|
||
- handler가 동적으로 만든 여러 output도 같은 transaction에 포함.
|
||
|
||
### MSG-019 — public API가 vendor 타입을 노출하지만 Gradle dependency는 숨기거나 지나치게 노출한다
|
||
|
||
**대표 근거**
|
||
|
||
- `CloudEventMapper` public API는 `io.cloudevents.CloudEvent`를 노출하지만 module dependency는
|
||
`implementation`이다.
|
||
- Kafka public constructor/mapper는 `Producer`, `ProducerRecord`, `RecordMetadata`를 노출하지만 Kafka client는
|
||
`implementation`이다.
|
||
- Rabbit public `RabbitChannelPublisher`는 Spring AMQP `Message`를 노출하지만 Spring AMQP는
|
||
`implementation`이다.
|
||
- `MessagingMetrics` public constructor는 Micrometer `MeterRegistry`, reactive facade는 Reactor `Mono`를
|
||
노출하지만 각각 implementation dependency다.
|
||
- 반대로 starter는 16개 internal module을 모두 `api`로 노출한다.
|
||
|
||
**구현 결정**
|
||
|
||
- 외부 public surface를 `api`와 의도한 SPI package로 allowlist한다. native mapper/coordinator/channel seam은
|
||
package-private 또는 `.internal`로 낮춘다.
|
||
- vendor extension이 의도된 public API라면 별도 native module에서 dependency를 `api`로 정확히 선언한다.
|
||
- 일반 internal project dependency는 `implementation`으로 낮추고 build-time consumer fixture로 검증한다.
|
||
- Revapi/japicmp 또는 repository public API snapshot과 `dev.caskeleton.messaging..` 전용 ArchUnit rule을
|
||
추가한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- Gradle TestKit 외부 Java consumer가 각 published artifact의 documented API만으로 compile.
|
||
- internal package import 실패, transitive Kafka/Rabbit이 core starter classpath에 없음.
|
||
|
||
### MSG-020 — codec/schema registry가 size limit 전에 전체 payload를 할당하고 version을 정본 key로 쓰지 않는다
|
||
|
||
**근거**
|
||
|
||
- JSON은 `JacksonMessageCodec.java:105-116`에서 `writeValueAsBytes` 후 size를 검사한다.
|
||
- Avro는 `AvroMessageCodec.java:96-113`에서 unbounded `ByteArrayOutputStream`/`toByteArray` 후 검사한다.
|
||
- Protobuf는 `ProtobufMessageCodec.java:82-87`에서 `toByteArray` 후 검사한다.
|
||
- Avro `decodeEvolved:163-179`는 normal decode와 달리 encoded length check가 없고 nested schema map은 outer
|
||
map만 copy한다.
|
||
- JSON/Protobuf registry는 `(MessageType, SchemaVersion)`이 아니라 message type만 key로 사용해 등록되지 않은
|
||
v999도 기존 class/parser로 decode하고 그 version label을 유지할 수 있다.
|
||
|
||
**영향**
|
||
|
||
configured max가 allocation bound가 아니므로 대형 object가 heap을 소진한 뒤에야 reject된다. schema
|
||
version과 실제 parser가 분리되면 compatibility gate와 audit가 거짓이 된다.
|
||
|
||
**구현 결정**
|
||
|
||
- max+1에서 즉시 예외를 내는 bounded `OutputStream`/coded stream을 사용하고 가능한 codec의 신뢰 가능한
|
||
serialized-size estimate도 먼저 검사한다.
|
||
- decode는 input bytes, depth, nesting, collection/string limits를 parser 전에/내부에서 모두 적용한다.
|
||
- registry key를 `MessageContractKey(MessageType, SchemaVersion)`로 바꾸고 descriptor가 Java class,
|
||
parser/schema, content type, compatibility policy를 함께 보유한다.
|
||
- construction 시 duplicate/mismatch를 fail-fast하고 Avro map을 deep immutable copy한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- max-1/max/max+1 및 훨씬 큰 streaming object에서 allocation/output이 limit 근처에서 중단.
|
||
- unregistered version, parser/class mismatch, duplicate registration 거절.
|
||
- Avro evolved oversized input과 caller nested map mutation.
|
||
- property/fuzz corpus: deep JSON, malformed varint, recursive/large collection schema.
|
||
|
||
### MSG-021 — admin approval은 위조 가능하고 one-shot 실행 기록이 process-local이다
|
||
|
||
**근거**
|
||
|
||
- `AdminApproval`, `ApprovedReplayPlan`, `ApprovedRedrivePlan`, destructive admin `Approved`는 public plain
|
||
constructor/record로 caller가 직접 승인 객체를 만들 수 있다.
|
||
- approval은 operation/source/target/plan digest/max impact에 cryptographically 또는 opaque capability로
|
||
bind되지 않아 다른 plan에 재사용될 수 있다.
|
||
- `AdminOperationIdempotencyStore.java:21-57`는 in-memory `ConcurrentHashMap`이고 starter도 이를 기본 등록한다.
|
||
- `DefaultMessagingAdminService.java:101-112,136-147`은 work 전에 ticket을 claim한다. 중간 실패하면 ticket은
|
||
소비됐지만 진행 위치/재개 상태가 없다.
|
||
- redrive loop의 synchronous failure는 뒤 item과 final audit을 건너뛸 수 있다.
|
||
|
||
**구현 결정: verified capability + durable operation journal**
|
||
|
||
- `ApprovalVerifier`만 issuer signature, subject separation, exact operation/source/target/topology version/plan
|
||
digest/max impact/expiry를 검증하고 opaque verified token을 만든다. 일반 caller가 verified type constructor를
|
||
호출할 수 없게 한다.
|
||
- shared DB journal에 unique `(approval_id, plan_digest)`와 `STARTED`, item progress, `COMPLETED`, `FAILED`,
|
||
lease/fencing을 저장한다.
|
||
- retry는 새 실행이 아니라 같은 operation을 checkpoint부터 resume한다. item별 redrive count/max enforcement와
|
||
audit를 transactionally 기록한다.
|
||
- destructive API와 non-destructive facade 모두 같은 approval authority/journal을 사용한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- 직접 constructor/서명 변조/wrong source-target/다른 plan 재사용/expired topology 거절.
|
||
- restart와 두 replica 경쟁에서 exactly one operation lease.
|
||
- N번째 item 실패 뒤 재개, max redrive count, audit failure/retry.
|
||
|
||
### MSG-022 — experimental adapter가 class-name 문자열로 오류를 추측하고 unknown을 definite rejection으로 낮춘다
|
||
|
||
**근거**
|
||
|
||
- Pulsar/NATS adapter는 exception class simple name substring으로 일부 오류를 분류한다.
|
||
- unknown send failure를 `REJECTED/not transmitted`로 두고 elapsed를 `Duration.ZERO`로 보고하며 per-call
|
||
timeout/options를 적용하지 않는다.
|
||
- NATS의 substring 판정은 unrelated exception 이름도 no-stream 계열로 오분류할 수 있다.
|
||
|
||
**영향**
|
||
|
||
experimental이더라도 caller가 `REJECTED`를 보고 새 ID로 재시도하면 실제 broker가 받은 message를
|
||
중복 생성할 수 있다. class name은 SDK version에 따라 바뀌는 비계약 문자열이다.
|
||
|
||
**구현 결정**
|
||
|
||
- 명시적인 typed SDK pre-send exception만 `REJECTED/NOT_TRANSMITTED`로 분류한다.
|
||
- write milestone 이후 또는 알 수 없는 failure의 기본값은 `AMBIGUOUS`다.
|
||
- adapter가 실제 client bridge/timeout/cancellation을 구현하기 전에는 `Extension` 또는
|
||
`Experimental contract seam`으로 문서화하고 Stable capability를 주장하지 않는다.
|
||
- contract test에 unknown subtype, wrapped exception, timeout, synchronous/asynchronous failure를 추가한다.
|
||
|
||
### MSG-023 — module 이름·폴더 구조와 canonical 정책 문서가 실제 43-leaf 구조를 설명하지 못한다
|
||
|
||
**근거**
|
||
|
||
- `messaging-outbox-jpa`/`messaging-inbox-jpa`는 JPA가 아니라 Spring JDBC + PostgreSQL 전용 SQL
|
||
(`?::jsonb`, `FOR UPDATE SKIP LOCKED`, `ON CONFLICT`, JSONB/BYTEA/TIMESTAMPTZ)을 사용한다.
|
||
- root policy는 19개 leaf라 선언하지만 registry/settings는 43개다.
|
||
- `src/messaging/CLAUDE.md`가 없어 API/policy/SPI/adapter/autoconfigure/testkit의 framework/public surface와
|
||
Stable promotion 규칙이 local authority에 없다.
|
||
- 신규 모든 leaf의 runtime membership이 empty지만 build-only/incubating 의미가 support matrix에 명확하지
|
||
않다.
|
||
|
||
**구현 결정**
|
||
|
||
- JDBC module은 `messaging-outbox-jdbc-postgresql`, `messaging-inbox-jdbc-postgresql`로 rename하고 vendor-neutral
|
||
port는 reliability API에 둔다.
|
||
- hard-coded leaf count를 registry에서 생성하거나 문서를 “registry가 소유하는 leaf 전체”로 표현하고
|
||
consistency task로 drift를 막는다.
|
||
- `src/messaging/CLAUDE.md`에 family별 허용 dependency, framework rule, public allowlist, stable evidence,
|
||
runtime membership/cutover 원칙을 추가한다.
|
||
- 24개 leaf를 무조건 합치지는 않는다. optional broker/codec isolation은 유지하되 아래 목표 tree처럼
|
||
family directory와 starter 경계를 명확히 한다.
|
||
|
||
### MSG-024 — 현재 사용 중인 legacy runtime에도 disabled/retry/payload/log 안전성 공백이 있다
|
||
|
||
**근거**
|
||
|
||
- `application.yml:466-476`과 `OutboxConfig.java:32-56`은 relay를 기본 활성화하지만 broker가 blank면
|
||
`MessagingConfig.java:40-46`이 disabled publisher를 만든다.
|
||
- 기존 relay `PublishPendingOutboxEventsUseCase.java:117-165`는 publish failure를 retry한 뒤 `DEAD`로
|
||
소진한다. 의도적으로 messaging-off인 환경에서도 pending row를 계속 claim할 수 있다.
|
||
- `application-core/.../OutboxEvent.java:36-48`은 payload non-null만 검사하고
|
||
`OutboxEnvelopeJson.java:18-40`은 raw payload를 전체 문자열에 그대로 삽입한다. JSON validity와 UTF-8
|
||
byte cap이 없다.
|
||
- `Slf4jOutboxRelayFailureReportAdapter.java:37-50`은 raw Throwable을 structured log cause로 전달한다.
|
||
allowlisted field와 별개로 exception message/stack에 payload, endpoint, secret이 포함될 수 있다.
|
||
- `OutboundMessagePublisher.java:27-35`는 broker send와 success logging을 한 try로 묶어, broker 성공 뒤
|
||
logger 예외를 publish failure로 오인할 수 있다.
|
||
- `KafkaAdapterSettings.java:17-26`은 trim한 값으로 regex 검증하지만 원문을 저장하고 port 0/99999를
|
||
허용한다.
|
||
|
||
**구현 결정**
|
||
|
||
- startup invariant `relay-enabled -> broker configured`를 강제하거나 disabled broker에서는 claim 자체를
|
||
중단해 PENDING을 보존한다.
|
||
- append/encode 경계에서 payload JSON을 strict parse/canonicalize하고 UTF-8 byte limit를 적용한다.
|
||
- failure report는 exception class + bounded sanitized code만 기록한다. raw Throwable을 운영 JSON log에
|
||
넣어야 한다면 최종 serialized output 전체에 검증된 redaction을 적용한다.
|
||
- publish outcome과 observation을 decorator로 분리해 logging failure가 broker result를 바꾸지 않게 한다.
|
||
- broker address를 typed host/port parser로 canonicalize하고 port 1..65535, bracketed IPv6를 검증한다.
|
||
|
||
**필수 테스트**
|
||
|
||
- broker blank + relay true startup failure 또는 PENDING 보존.
|
||
- malformed JSON, max bytes ±1, Unicode/surrogate/large payload.
|
||
- 실제 Logstash encoder JSON에 arbitrary secret/payload/endpoint 문자열 없음.
|
||
- broker confirmed + logger throw에서도 application publish 성공 유지.
|
||
- host whitespace, port 0/65535/65536, IPv6.
|
||
|
||
### MSG-025 — 운영 문서와 outbox reclaim 설명이 실제 코드에서 drift했다
|
||
|
||
**근거**
|
||
|
||
- `docs/messaging/outbox-inbox.md:41-48`은 claim 대상을 `PENDING`, `AMBIGUOUS`로만 설명하지만 구현과
|
||
partial index는 lease-expired `IN_FLIGHT`도 reclaim한다.
|
||
- `docs/runbooks/outbox-publish-failed.md:28`의 `APP_MESSAGING_KAFKA_ENABLED`와 문서의 adapter class 이름은
|
||
현재 env/class와 맞지 않는다.
|
||
- configuration reference의 root prefix와 실제 binder 불일치는 MSG-008에 별도로 다뤘다.
|
||
|
||
**구현 결정/테스트**
|
||
|
||
- runbook 명령/env/class/predicate를 executable documentation test 또는 source-generated snippet으로
|
||
연결한다.
|
||
- relay crash → lease expiry → same-ID reclaim integration test 이름을 문서 evidence에 링크한다.
|
||
- 문서의 Stable 문구는 MSG-014 evidence manifest가 존재할 때만 생성/승격한다.
|
||
|
||
## 6. 권장 목표 아키텍처와 폴더 구조
|
||
|
||
### 6.1 책임 흐름
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
A[application-owned messaging port] --> B[platform anti-corruption bridge]
|
||
B --> C[DefaultMessagePublisher]
|
||
C --> D[configuration/profile compiler]
|
||
C --> E[codec + canonical envelope codec]
|
||
C --> F[security + admission permit]
|
||
C --> G[runtime generation lease]
|
||
G --> H{MessagingTransport Strategy}
|
||
H --> K[Kafka adapter]
|
||
H --> R[Rabbit adapter]
|
||
H --> X[Experimental adapters]
|
||
K --> O[typed PublishResult]
|
||
R --> O
|
||
X --> O
|
||
O --> P[application PublicationOutcome]
|
||
|
||
K2[Kafka delivery] --> Q[DefaultDeliveryProcessor]
|
||
R2[Rabbit delivery] --> Q
|
||
Q --> M[MessageHandler]
|
||
M --> N[HandleResult + RetryDecision]
|
||
N --> L[confirmed DLQ / source settlement]
|
||
Q --> I[transactional inbox + business Unit of Work]
|
||
```
|
||
|
||
핵심은 application 경계, platform orchestration, broker strategy, persistence adapter를 분리하면서도
|
||
publish/consume 각각의 **한 개짜리 실행 pipeline**을 두는 것이다. 지금처럼 validator, limiter, runtime
|
||
registry, transport, observation을 bean으로만 제공하면 호출 순서와 누락을 보장할 수 없다.
|
||
|
||
### 6.2 권장 tree
|
||
|
||
아래는 최종 방향이다. 한 번에 물리 이동하지 말고 behavior fix와 compatibility test 뒤 별도 change set으로
|
||
진행한다.
|
||
|
||
```text
|
||
src/
|
||
├── application-core/.../messaging/
|
||
│ ├── event/ # business semantic draft/validated event
|
||
│ ├── publication/ # application-owned port/outcome
|
||
│ └── reliability/ # application transaction policy
|
||
├── adapter/outbound/messaging/
|
||
│ ├── platformbridge/ # temporary anti-corruption layer
|
||
│ └── legacy/ # cutover 동안 신규 기능 금지
|
||
├── messaging/
|
||
│ ├── CLAUDE.md
|
||
│ ├── api/
|
||
│ │ ├── messaging-core-api/
|
||
│ │ ├── messaging-schema-api/
|
||
│ │ └── messaging-reliability-api/
|
||
│ ├── runtime/
|
||
│ │ ├── messaging-runtime-core/ # publisher/delivery pipelines
|
||
│ │ ├── messaging-policy/
|
||
│ │ └── messaging-transport-spi/
|
||
│ ├── codec/
|
||
│ │ ├── json/
|
||
│ │ ├── avro/
|
||
│ │ ├── protobuf/
|
||
│ │ └── cloudevents/
|
||
│ ├── adapter/
|
||
│ │ ├── kafka/
|
||
│ │ ├── rabbit/
|
||
│ │ └── experimental/{kafka-share,pulsar,nats}/
|
||
│ ├── reliability/
|
||
│ │ ├── outbox-jdbc-postgresql/
|
||
│ │ ├── inbox-jdbc-postgresql/
|
||
│ │ └── claim-check/
|
||
│ ├── support/
|
||
│ │ ├── observability-micrometer/
|
||
│ │ └── security/
|
||
│ ├── admin/{api,runtime}/
|
||
│ ├── spring/
|
||
│ │ ├── autoconfigure-core/
|
||
│ │ └── starter/{core,kafka,rabbit,reliability-jdbc-postgresql,admin}/
|
||
│ └── testkit/
|
||
└── app-bootstrap/.../messaging/
|
||
└── PlatformMessagingComposition.java
|
||
```
|
||
|
||
### 6.3 visibility/dependency 규칙
|
||
|
||
- application은 신규 broker/runtime type을 직접 보지 않고 자기 port만 소유한다.
|
||
- `api`/의도한 `spi` package만 public이다. mapper/coordinator/native bridge/auto-config helper는 internal이다.
|
||
- broker/codec optionality를 위해 leaf 분리는 유지한다. 두 broker를 common starter가 의존하지 않는다.
|
||
- runtime-core는 broker adapter에 의존하지 않고 `MessagingTransport` strategy만 본다.
|
||
- broker adapter는 runtime-core의 concrete class를 import하지 않고 SPI와 자기 native SDK만 본다.
|
||
- PostgreSQL reliability adapter는 vendor 특성을 숨기지 않되 JDBC `Connection`은 core port 밖으로 유출하지
|
||
않는다.
|
||
- testkit의 expected contract와 certification artifact를 분리한다. 기대 matrix 자체가 실행 evidence가
|
||
되어서는 안 된다.
|
||
|
||
## 7. 디자인 패턴 적용 판단
|
||
|
||
패턴은 이름을 늘리기 위해서가 아니라 현재 깨진 불변식을 한곳에서 강제하기 위해 사용한다.
|
||
|
||
| 문제 | 적용할 패턴 | 구체 적용 | 피할 것 |
|
||
|---|---|---|---|
|
||
| publish 단계 누락 | Facade + ordered Pipeline/Decorator | `DefaultMessagePublisher`, typed steps, finally permit/lease release | 자유 순서 interceptor/service locator |
|
||
| broker 차이 | Strategy + Abstract Factory | `MessagingTransport`, broker별 `BrokerRuntimeFactory` | Kafka/Rabbit/NATS를 상속 Template Method로 평탄화 |
|
||
| profile 검증 | Strategy + Composite | generic/broker/security rule을 compiled graph에 적용 | bean만 만들고 호출자에게 검증 책임 전가 |
|
||
| outbox/inbox 원자성 | Unit of Work | transaction-aware JDBC adapter + application transaction boundary | core port에 `Connection` 노출, after-commit append |
|
||
| lease/offset/confirm | State Machine + Fencing Token | outbox token CAS, assignment epoch, Rabbit confirm events | boolean/status 임의 update |
|
||
| invalid result 방지 | Sealed ADT + private factory | publish/settlement/admin verified token | 모든 조합을 받는 public record constructor |
|
||
| backpressure/lifecycle | AutoCloseable Permit | destination/work/runtime/credential lease exactly-once close | 문자열 기반 release/end 호출 |
|
||
| envelope 정합성 | Canonical Mapper/Codec | Kafka/Rabbit/outbox/CDC 공통 `EnvelopeHeaderCodec` | adapter마다 reserved header 재구현 |
|
||
| 기존→신규 migration | Anti-Corruption Layer | application outcome/wire mapping과 single authority | 두 hierarchy/두 relay의 영구 병행 |
|
||
| retry/DLQ/admin | durable State Machine/Journal | next-at/park, confirmed-DLQ-before-ACK, resumable admin | memory-only counter/approval claim |
|
||
|
||
다음은 도입하지 않는 편이 낫다.
|
||
|
||
- broker별 confirmation/settlement 의미가 다른데 거대한 `AbstractBrokerTransport` base class로 합치기
|
||
- PostgreSQL lock/isolation을 숨기는 범용 repository와 모든 DB를 지원한다고 보이게 만들기
|
||
- 외부 연산이 단순한 sealed result에 별도 Visitor class hierarchy를 과도하게 추가하기
|
||
- exception class-name substring classifier, global mutable plugin registry, raw broker hint map
|
||
- outbox만으로 exactly-once를 약속하는 facade
|
||
|
||
## 8. 구현 순서
|
||
|
||
각 wave는 독립 merge/review 가능한 범위다. 뒤 wave가 앞 wave의 안전성 gate를 우회하지 않게 한다.
|
||
|
||
### Wave 0 — 사실성 및 회귀 테스트를 먼저 고정
|
||
|
||
1. 신규 platform을 `Contract-only/Build-only`로 표시하고 Stable promotion/cutover를 보류한다.
|
||
2. MSG-001/002/004/005/016/018 재현 테스트를 먼저 실패하도록 추가한다.
|
||
3. support matrix의 4.2/4.3 및 Rabbit 5-fault live claim을 실제 evidence 수준으로 낮춘다.
|
||
4. legacy/new canonical model과 single publication authority ADR을 작성한다.
|
||
|
||
완료 기준: CI와 문서가 현재 executable evidence보다 강한 주장을 하지 않고, 데이터 유실 시나리오가
|
||
red test로 재현된다.
|
||
|
||
### Wave 1 — transactional reliability와 durable state
|
||
|
||
1. `JdbcOutboxRepository`/`JdbcInboxRepository`를 transaction-aware JDBC로 교체한다.
|
||
2. interface/`IdempotentConsumer` 경로의 PostgreSQL commit/rollback IT를 추가한다.
|
||
3. outbox V2 migration에 lease owner/token/next-at/parked state를 추가한다.
|
||
4. lease-bearing port와 CAS terminal transition을 구현한다.
|
||
5. retry scheduler/maxAttempts를 relay와 DB claim에 연결하고 purge를 bounded SQL로 바꾼다.
|
||
6. 가능하면 이미 더 강한 application-core v2 delivery owner/CAS pattern을 canonical로 재사용한다.
|
||
|
||
완료 기준: business row와 inbox/outbox가 원자적으로 움직이고 stale relay가 어떤 순서에서도 최신 state를
|
||
바꾸지 못한다.
|
||
|
||
### Wave 2 — broker state machine
|
||
|
||
1. Kafka poll batch partition별 rewind/dispatch와 executor rejection 보상을 구현한다.
|
||
2. assignment epoch, revoke drain, poll-thread command queue, post-commit watermark를 구현한다.
|
||
3. Kafka decode/handler/DLQ/settlement outcome을 분리하고 handler timeout을 적용한다.
|
||
4. Rabbit native channel bridge, multiple confirm, return correlation, deadline, close drain을 구현한다.
|
||
5. Rabbit consumer/settlement를 typed state machine과 central delivery processor로 연결한다.
|
||
6. Kafka transactional processor callback 순서를 바로잡는다.
|
||
|
||
완료 기준: fault/rebalance/timeout/close 모든 테스트에서 미처리 source가 commit/ACK되지 않고 pending future와
|
||
permit이 0으로 돌아온다.
|
||
|
||
### Wave 3 — runtime composition과 starter
|
||
|
||
1. `messaging-runtime-core`와 `DefaultMessagePublisher`/`DefaultDeliveryProcessor`를 추가한다.
|
||
2. canonical `app.messaging` settings compiler와 composite validator를 구현한다.
|
||
3. broker runtime factory를 구현해 native producer/consumer/security/lifecycle을 완결한다.
|
||
4. core/Kafka/Rabbit/reliability/admin starter를 분리하고 `api` exposure를 축소한다.
|
||
5. `SmartLifecycle` drain, generation sweep, admission/work/runtime/credential permit을 pipeline에 연결한다.
|
||
|
||
완료 기준: documented profile 하나만으로 full context가 fake facade 없이 actual transport bean까지 조립되고,
|
||
불완전 profile은 worker 시작 전에 실패한다.
|
||
|
||
### Wave 4 — canonical wire/schema/security
|
||
|
||
1. `EnvelopeHeaderCodec`과 application/platform header type을 도입한다.
|
||
2. outbox schema에 canonical metadata를 보존하고 polling/CDC mapper를 통합한다.
|
||
3. `(MessageType, SchemaVersion)` schema registry와 bounded codec output/decode를 구현한다.
|
||
4. PublishOptions normalization, runtime capability, sealed PublishResult를 구현한다.
|
||
5. identifier/header/trace/key wire validation, TLS allowlist, JAAS escaping을 적용한다.
|
||
6. metrics tag allowlist와 structured diagnostic event를 분리한다.
|
||
|
||
완료 기준: direct/Kafka/Rabbit/outbox polling/CDC round-trip이 같은 canonical envelope를 만들며 malicious wire
|
||
input이 broker/DB/metric backend 전에 거절된다.
|
||
|
||
### Wave 5 — application bridge와 cutover
|
||
|
||
1. application-owned publication outcome을 확장하고 platform bridge를 구현한다.
|
||
2. golden wire/semantic compatibility와 dual-authority negative test를 추가한다.
|
||
3. 기존 outbox writer/storage를 유지한 transport-only cutover를 한다.
|
||
4. mixed-version soak 후 별도 release에서 storage/consumer/inbox authority를 이동한다.
|
||
5. old backlog, generation, retry/DLQ, retention window가 drain된 뒤 legacy API/config/table을 제거한다.
|
||
|
||
완료 기준: rolling deployment와 rollback에서 한 business fact당 writer/relay/publish authority가 항상 하나다.
|
||
|
||
### Wave 6 — admin 및 release qualification
|
||
|
||
1. verified approval token과 durable resumable admin journal을 구현한다.
|
||
2. broker/version/scenario evidence manifest와 fail-closed certification task/workflow를 추가한다.
|
||
3. TLS/SASL/ACL, broker failover, multi-node, backlog, soak, p95/p99/allocation 기준을 실행한다.
|
||
4. evidence가 있는 조합만 Stable로 승격한다.
|
||
|
||
## 9. 권장 테스트 구조와 명령
|
||
|
||
### 9.1 새 테스트 lane
|
||
|
||
```text
|
||
messaging-runtime-core:test
|
||
- full publish/delivery pipeline transition tables
|
||
- option/capability/result invariants
|
||
|
||
messaging-reliability-postgresql-test
|
||
- public port UnitOfWork rollback
|
||
- two-relay fencing/retry/purge
|
||
|
||
messaging-kafka:test
|
||
- partition/epoch/commit/executor deterministic tests
|
||
- transaction callback tests
|
||
|
||
messaging-rabbit:test
|
||
- confirm/return/timeout/close state machine
|
||
- consumer/DLQ settlement tests
|
||
|
||
messaging-wire-contract-test
|
||
- Kafka/Rabbit/outbox/CDC canonical round-trip
|
||
- adversarial header/schema/trace/property tests
|
||
|
||
messagingCertificationTest
|
||
- broker-version matrix
|
||
- five fault scenarios
|
||
- Docker required, zero skip, evidence manifest
|
||
|
||
messagingCutoverTest
|
||
- legacy bridge golden bytes
|
||
- mixed-version authority switch/rollback
|
||
```
|
||
|
||
### 9.2 구현 중 focused 검증
|
||
|
||
```bash
|
||
cd src
|
||
./gradlew :messaging:messaging-core-api:test --console=plain
|
||
./gradlew :messaging:messaging-transport-spi:test --console=plain
|
||
./gradlew :messaging:messaging-outbox-jpa:test --console=plain
|
||
./gradlew :messaging:messaging-inbox-jpa:test --console=plain
|
||
./gradlew :messaging:messaging-kafka:test --console=plain
|
||
./gradlew :messaging:messaging-rabbit:test --console=plain
|
||
./gradlew :messaging:messaging-spring-boot-starter:test --console=plain
|
||
./gradlew :adapter:outbound:messaging:test --console=plain
|
||
./gradlew verifyCleanArchitectureDependencies verifyRuntimeModuleMembership --console=plain
|
||
```
|
||
|
||
module rename/split 뒤에는 registry의 `gradle_path`에서 새 명령을 파생하고 위 이름을 함께 갱신한다.
|
||
|
||
### 9.3 release 전 필수 검증
|
||
|
||
```bash
|
||
cd src
|
||
./gradlew messagingCertificationTest --console=plain
|
||
./gradlew messagingCutoverTest --console=plain
|
||
./gradlew test --console=plain
|
||
./gradlew check --console=plain
|
||
```
|
||
|
||
새 task는 source set과 executable suite를 함께 추가한다. 문서에 존재하지 않는 task 이름만 미리 약속하지
|
||
않는다. certification lane은 Docker/broker/version/test가 없으면 skip이 아니라 failure다.
|
||
|
||
## 10. 완료 정의
|
||
|
||
다음 질문에 모두 코드, DB state, broker evidence, CI artifact로 “예”라고 답할 수 있을 때만 신규 messaging
|
||
platform을 runtime-ready Stable로 판정한다.
|
||
|
||
- public outbox/inbox port가 실제 application transaction과 같은 resource에서 commit/rollback하는가?
|
||
- lease-expired worker, duplicate callback, delayed confirm이 최신 terminal state를 덮어쓸 수 없는가?
|
||
- retry budget/next-at/parking/retention delete가 restart와 다중 replica에서도 durable한가?
|
||
- Kafka poll batch의 모든 미제출 record가 처리 또는 rewind되고 assignment epoch가 stale ACK를 막는가?
|
||
- Kafka offset local state는 broker commit 성공 뒤에만 전진하는가?
|
||
- Rabbit multiple confirm, return, NACK, timeout, send throw, channel close가 모든 future를 정확히 종료하는가?
|
||
- decode/handler/DLQ/settlement failure가 분리되고 confirmed DLQ 전 source ACK/commit이 금지되는가?
|
||
- documented configuration만으로 publisher/consumer/runtime이 조립되며 invalid/unknown 설정이 fail-closed인가?
|
||
- timeout/confirmation/dedup/hint capability가 실제 runtime behavior와 일치하는가?
|
||
- tenant/trace/schema/application header/retry/key가 direct/polling/CDC/Kafka/Rabbit에서 동일하게 보존되는가?
|
||
- credential/TLS/auth rotation 중 사용 중인 secret/client가 조기 clear/close되지 않는가?
|
||
- metric tag와 log 어디에도 unbounded identifier/secret/payload가 들어가지 않는가?
|
||
- application과 platform 사이 writer/relay authority가 한 시점에 정확히 하나인가?
|
||
- admin approval을 caller가 위조/다른 plan에 재사용할 수 없고 restart 뒤 안전하게 재개되는가?
|
||
- support matrix의 각 Stable broker/version/scenario가 같은 commit의 zero-skip evidence에 연결되는가?
|
||
- root/local architecture policy와 registry/module count/runtime membership이 모순되지 않는가?
|
||
|
||
하나라도 아니면 해당 기능은 `Experimental`, `Contract-only`, `Unwired` 중 실제 상태로 표시한다.
|
||
|
||
## 11. 이번 리뷰에서 실행한 검증
|
||
|
||
검토 중 HEAD가 messaging merge commit `71c0d2122f2c65e9ce7910c6615b57056d9cebb6`에서
|
||
`c3043e530a604315c4df341b87b5470c7617ea03`으로 이동했다. 두 commit 사이 변경은 GraphQL module에
|
||
한정됐고 messaging/application/bootstrap/registry/build 통합 경로에는 변경이 없었다. `src/messaging`
|
||
tree hash도 `20664539b0609c6c413759e2b2945bf421c10de7`로 동일했다. 그 뒤 최종 HEAD에서 production code를
|
||
수정하지 않은 채 신규/legacy test와 architecture gate를 `--rerun-tasks`로 다시 실행하고 JUnit XML을
|
||
별도로 합산했다.
|
||
|
||
| 명령 | 결과 | 관측 범위 |
|
||
|---|---|---|
|
||
| 신규 24개 messaging leaf의 모든 `:test` + legacy `:test` + architecture/runtime/one-type gate, `--rerun-tasks --no-daemon --max-workers=2` | BUILD SUCCESSFUL, 3m 53s, 107/107 tasks executed | 최신 HEAD의 신규 600 tests + legacy 81 tests, failure/error/skip 0; 43-leaf gate 통과 |
|
||
| `./gradlew :adapter:outbound:messaging:test :messaging:messaging-transport-spi:test :messaging:messaging-outbox-jpa:test :messaging:messaging-inbox-jpa:test :messaging:messaging-spring-boot-starter:test :messaging:messaging-testkit:test --rerun-tasks --console=plain` | BUILD SUCCESSFUL, 1m 27s, 61/61 tasks executed | 핵심 reliability/runtime/starter/testkit fresh 재확인 |
|
||
| `./gradlew verifyCleanArchitectureDependencies verifyRuntimeModuleMembership verifyOneTypePerFile --rerun-tasks --console=plain` | 위 최신-HEAD 통합 실행에 포함되어 BUILD SUCCESSFUL | 현재 43-leaf registry dependency/runtime/one-type gate |
|
||
| 신규 24 leaf와 legacy adapter의 Checkstyle/SpotBugs, legacy JSON runtime graph, `verifyDependencyLocks verifyCleanArchitectureDependencies --console=plain --no-daemon --max-workers=2` | 최신 HEAD에서 BUILD SUCCESSFUL in 6m, 223 tasks(96 executed/127 up-to-date) | 신규/legacy 정적 분석, 전체 dependency locks, architecture |
|
||
|
||
신규 600개 집계에는 다음 Docker-backed IT가 실제 발견·실행됐고 skip은 0이었다.
|
||
|
||
- `InboxPostgresIT`, `OutboxPostgresIT`
|
||
- `KafkaBrokerIT`, `KafkaAmbiguityChaosIT`, `KafkaConsumerSettlementIT`, `KafkaReadCommittedIT`
|
||
- `KafkaTopologyValidationIT`, `KafkaTransactionFencingIT`, `KafkaTransactionIT`
|
||
- `RabbitBrokerIT`
|
||
|
||
테스트가 모두 통과했는데도 P0/P1이 남는 이유는 명확하다. transaction rollback IT는 production port가
|
||
아닌 safe `Connection` overload를 호출하고, full starter imports는 fake publisher 없이 로드하지 않으며,
|
||
failure matrix는 실행 artifact가 아닌 hard-coded map을 검사한다. 현재 test green을 production composition
|
||
green으로 해석하면 안 된다.
|
||
|
||
### 11.1 실패 및 미실행 검증
|
||
|
||
- 최신 HEAD에서 다시 실행한 `./gradlew :adapter:outbound:messaging:check --console=plain --no-daemon
|
||
--max-workers=2`는 `BUILD FAILED in 23s`였다. messaging assertion 실패가 아니라 root 선행
|
||
`verifyNoStaleTraceableJars`가 현재 `c3043e530a60` archive와 함께 과거 hash JAR을 보유한 archive task
|
||
31개를 발견해 중단했다. core/shared/sample, 신규 messaging 17개 leaf, inbound web, outbound
|
||
cache/fileserver/httpclient/identifier/legacy-messaging/notification/objectstorage/persistence-jpa/support가
|
||
대상이다. 사용자 build artifact를 임의 삭제하지 않기 위해 `cleanStaleTraceableJars`는 실행하지 않았다.
|
||
- 최초 sandbox Gradle 시도는 user Gradle cache의 lock 파일이 read-only라 실패했고, 같은 명령을 승인된
|
||
Gradle 실행으로 재실행해 위 성공 결과를 얻었다.
|
||
- repository 전체 `test`/`check`, messaging JMH, 외부 dependency vulnerability DB/Trivy는 이번 review
|
||
범위에서 실행하지 않았다.
|
||
- Kafka 4.2/4.3, Rabbit 5개 network fault 전체, TLS/SASL/ACL, multi-node failover, process restart/cutover,
|
||
broker load/soak test lane은 현재 없거나 실행하지 않았다.
|
||
- `messaging-reliability-api:test`는 `NO-SOURCE`다. public reliability record/port contract의 직접 테스트는
|
||
adapter 간접 테스트 외에 추가할 필요가 있다.
|
||
|
||
위 architecture/static gate의 통과는 현재 registry가 스스로 일관된다는 증거다. root policy의 “19개”와
|
||
실제 43개가 의미적으로 일치하거나 신규 platform이 runtime-ready라는 증거는 아니다.
|
||
|
||
## 12. LLM Wiki 캡처
|
||
|
||
필수 작업 기록은 `/home/donghyeon/workspace/ai-tool/llm-wiki-private/raw/branch-notes/main.md`의
|
||
`2026-08-14 캡처 — merge 이후 Messaging 모듈 상세 리뷰` 섹션에 갱신했다. 검토 범위, 25개 finding의
|
||
핵심 판정, 구현 순서, 변경 파일, 실행한 검증, 실패·미실행 범위와 evidence grade를 기록했다.
|
||
|
||
이번 작업은 production 구현 전 read-only review이므로 `raw/errors/`, `raw/interviews/`,
|
||
`raw/blog-topics/` 및 canonical `wiki/` 파생 문서는 만들지 않았다. 실제 P0/P1 구현과
|
||
transaction/rebalance/restart/cutover evidence가 생긴 뒤 파생 여부를 다시 판단한다.
|