document-haness/docs/clean-architecture-backend-template/analysis/messaging/messaging-kafka.md

# messaging-kafka 완전 해부

> 상태: COMPLETE
> 재오픈 게이트: cycle 2 재통독(2026-09-01) — `src/main` production 34파일 3,427줄 + `src/test` 24파일 4,087줄 축자 통독 완료. `STRUCTURAL_ONLY` 는 `gradle.lockfile` 하나.
> 기준 revision: `21234e38cdb9a926cbc92bb97a2aee2e4a7d2916`
> 분석 범위: `src/messaging/messaging-kafka`
> SSOT owner: `messaging-kafka`
> integration/family document: `analysis/19-messaging-platform.md` (secondary, INTEGRATION_ONLY)

---

## 0. SSOT identity / 커버리지

- `runtime_memberships`: **`["app-bootstrap"]`** — 출하
- 등급: Stable · 이 가족에서 실제로 선택 가능한 유일한 브로커(§12.1)

| 파일 | LOC | 조립되나 |
|---|---:|---|
| `KafkaConsumerRegistrar` | 621 | **아니오** — 테스트만 |
| `KafkaMessagingTransport` | 215 | 예(발행 전용 생성자) |
| `KafkaTransactionalPublisher` | 178 | 아니오 |
| `KafkaSecurityConfigurer` | 173 | 빈으로만 — 호출처 없음 |
| `KafkaHeaderMapper` | 169 | 예(전송 경유) |
| `KafkaDeliveryMapper` | 167 | 아니오 |
| `KafkaBatchConsumerRegistrar` | 146 | **아니오** — 저장소 전체에 참조 0 |
| `PartitionWorkCoordinator` | 128 | 아니오 |
| `KafkaRetryExecutor` | 126 | 아니오 |
| `ContiguousPartitionOffsetTracker` | 119 | 아니오 |
| `KafkaPublishFailureClassifier` | 116 | 예(전송 경유) |
| `KafkaPublishMapper` | 111 | 예(전송 경유) |
| `KafkaRetryMetadataMapper` | 93 | 아니오 |
| `KafkaPartitionRetryScheduler` | 90 | 아니오 |
| `KafkaReplayCapability` | 82 | 아니오 |
| `KafkaRetryTopicPublisher` | 75 | 아니오 |
| `SpringKafkaTransactionalProcessor` | 68 | 아니오 |
| `KafkaOffsetResetExecutor` | 63 | 아니오 |
| `KafkaProfileValidator` | 60 | 예(시작 검증) |
| `KafkaReplayPlanner` | 58 | 아니오 |
| `KafkaBrokerProfile` | 56 | 예(설정 컴파일) |
| `KafkaSettlementQueue` · `KafkaTransactionProfileValidator` | 52 · 52 | 아니오 / 빈만(§17.2) |
| `KafkaSettlementCommand` | 50 | 아니오 |
| `KafkaDeadLetterPublisher` | 49 | 아니오 |
| `PartitionOffsetTracker` | 48 | 아니오 |
| `KafkaTopologyInspector` | 45 | 아니오 |
| `KafkaRetryOutcome` | 40 | 아니오 |
| `KafkaPosition` | 38 | 예(발행 결과) |
| `KafkaReplayPlan` | 36 | 아니오 |
| `KafkaQuarantinePublisher` | 31 | 기본 구현만 |
| `KafkaTransactionalProcessor` | 29 | 아니오 |
| `KafkaTransactionalDelivery` · `KafkaTransactionalOutput` | 22 · 21 | 아니오 |

main 총 **34파일 / 3,427줄**.

### Coverage ledger

| scope | count | disposition | reason |
|---|---:|---|---|
| `main/java/**` | 34 | `FULL_READ` | 3,427줄. 위 표가 전부 |
| `test/java/**` | 24 | `FULL_READ` | 4,087줄. 인증 레인·Toxiproxy 레인 포함 |
| `build.gradle` | 1 | `FULL_READ` | 전문 |
| `gradle.lockfile` | 1 | `STRUCTURAL_ONLY` | 잠금 파일 — 생성물 |

`UNCLASSIFIED` 0.

> 이 표는 2026-09-01 재통독에서 다시 세었다. 이전 판은 `main/java/**` 를 **29** 로 적었다. 실제는 34 이고, 빠져 있던 다섯 안에 §17.4 의 `KafkaRetryMetadataMapper` 가 있었다. "조립되나" 열도 이번에 추가했다 — 이 리프의 판정 등급이 전부 그 열에 달려 있다.

---

## 1. 소비자 런타임 — 스레드 규율이 설계다

> "Every call into `Consumer` — poll, pause, resume, seek, commit — happens on the poll thread and
> nowhere else. `KafkaConsumer` is documented as not thread-safe, and a worker that committed
> directly would corrupt the client's internal state under concurrency in ways that surface much
> later as skipped offsets. Workers therefore enqueue a `KafkaSettlementCommand` and the poll thread
> applies it at the top of the next cycle."

공개 API 인 `pause`/`resume` 도 제어 큐를 통해 폴 스레드로 넘어가고, 반환된 단계는 **다음 폴 주기** 에 완료된다.

> "so a caller that awaits it knows the consumer is paused rather than merely asked to pause… there
> is no safe way to touch the consumer from another thread, so 'paused' cannot be true until the loop
> says so."

`close()` 만 예외이고 그 예외에 근거가 붙어 있다 — 이후 폴 루프가 멈추므로 큐에 넣으면 영원히 배수되지 않는다.

이 규율은 실제로 지켜진다. 작업자 람다가 만지는 것은 `settlements`·`coordinator`·`shutdown`·`retries` 뿐이고 `consumer` 는 한 번도 없다. 통독으로 확인했다.

## 2. 커밋은 연속 워터마크로만 전진한다

> "A Kafka offset commit is a watermark, not a set: committing offset 13 declares that everything
> below it is done. With concurrent handlers, offsets finish out of order — 10 and 12 may complete
> while 11 is still running — and committing 13 at that moment would silently discard 11."

그 대가도 적혀 있다 — 느린 메시지 하나가 그 파티션의 워터마크를 붙든다. 그것이 옳은 교환이라는 근거는 대안이 메시지를 잃는다는 것이고, 지연은 소비자 랙으로 보인다는 것이다.

그리고 등록만 되고 제출되지 않은 오프셋을 되돌리는 경로가 있다.

> "A delivered offset with no worker behind it holds the contiguous watermark back forever: nothing
> will ever complete it, so the partition stops committing while continuing to consume."

## 3. 이미 고쳐진 결함 네 개가 코드에 주석으로 남아 있다

이 리프의 서술 방식이다 — 고친 자리마다 이전 상태를 적어 둔다.

**커밋 순서.** 지역 맵과 트래커를 `commitSync` **뒤에** 갱신한다. 이전 순서는 실패한 커밋 뒤에 브로커가 받은 적 없는 오프셋을 커밋된 것으로 믿게 했고, 잘린 트래커가 그것을 다시 만들 수 없어 다음 커밋이 간극을 건너뛰었다.

**재조정 에폭.** 파티션 회수 시 에폭을 **먼저** 지운다.

> "Any settlement still in flight for these partitions now carries a number no live assignment has,
> so applySettlements refuses it instead of moving a watermark this consumer no longer owns."

**전역 break 제거.** 한 파티션이 천장에 닿았을 때 배치 전체를 버리던 형태를 파티션별 처리로 바꿨다.

> "which abandoned every record the same poll had returned for *other* partitions… a processing gap
> that nothing reported."

**정착의 단일 종결.** `acknowledge`/`requeue`/`discard` 가 하나의 CAS 를 두고 경쟁한다 — 핸들러가 둘 다 말하면 폴 스레드가 두 번째를 믿던 형태를 막는다.

거부된 정착 수는 조용히 세지 않고 `staleSettlements()` 로 노출한다. 0 이 아니면 핸들러가 자기 할당보다 오래 살고 있다는 뜻이고, 운영자가 행동할 수 있는 신호다.

## 4. 배압은 버퍼가 아니라 일시정지로 준다

> "A partition at its in-flight ceiling stops being fetched, so unprocessed records stay in the
> broker instead of in the heap."

파티션 단위로만 멈추고, 재개 지점은 그 파티션의 가장 이른 미제출 오프셋이다. 재개 자체가 없는 것이 §17.3 이다.

## 5. 발행 실패 분류

> "The split is between failures that prove the record was not stored and failures that prove
> nothing… The default is deliberately ambiguous rather than rejected. Guessing 'rejected' on an
> unknown error is what turns one lost confirmation into two orders."

일곱 예외 타입만 단정적 거부이고, 그중 셋(`AuthenticationException`·`AuthorizationException`·`SerializationException`)은 각각 전용 범주로 간다.

## 6. 트랜잭션 조건

`KafkaTransactionProfileValidator` 가 넷을 요구한다 — 트랜잭션 식별자 접두, 멱등 생산자, `acks=all`, 수동 오프셋 커밋. 그리고 다섯째가 핵심이다.

> "A destination that declares `INBOX_TRANSACTIONAL` is telling the platform its side effect lives in
> a database, and a Kafka transaction cannot span that. Allowing both to be configured together would
> let a team read 'transactional' twice and conclude the whole path is atomic when the two halves can
> still diverge."

## 10. 테스트 레인

24파일 4,087줄. 세 층이다.

| 층 | 파일 | 무엇을 붙드나 |
|---|---|---|
| 결정적 | `KafkaConsumerRegistrarTest`(440), `KafkaTransactionOrderingTest`(192), `KafkaProfileValidatorTest`(167), `KafkaEnvelopeRoundTripTest`(154), `ReservedHeaderForgeryTest`(123), `KafkaHeaderMapperTest`(118), `ContiguousPartitionOffsetTrackerTest`(94), `KafkaReplayPlannerTest`(88), `PartitionWorkCoordinatorTest`(77), `KafkaProducerContractTest`(24) | `MockConsumer`·`MockProducer` 로 폴 주기·트랜잭션 호출 순서·헤더 왕복·워터마크 산술 |
| 실브로커 IT | `KafkaBrokerIT`(292), `KafkaConsumerSettlementIT`(218), `KafkaAmbiguityChaosIT`(173), `KafkaTransactionIT`(168), `KafkaReadCommittedIT`(155), `KafkaTransactionFencingIT`(150), `KafkaTopologyValidationIT`(144), `KafkaContainerSmokeTest`(66) | Testcontainers `apache/kafka:4.1.0` |
| 인증 레인 | `KafkaBrokerCertificationIT`(466) | Toxiproxy 로 소켓 단위 결함 주입 |

인증 레인의 판단이 이 가족에서 가장 강하다.

> "No `@EnabledIf` on Docker, deliberately… a certification lane that skips reports success for a
> broker nobody started, which is the exact failure the evidence exists to rule out."

그리고 커버하지 못하는 시나리오를 숨기지 않는다 — `connection-refused` 는 Kafka 생산자가 연결 성립 전에 레코드를 버퍼링하므로 전송에 대해 아무것도 증명하지 못하는 배달 마감으로만 나타난다. 그래서 그것을 `knownGaps` 로 남긴다.

`KafkaReadCommittedIT.abortATransactionCarrying` 의 주석도 같은 종류다 — `flush()` 가 없으면 abort 가 클라이언트 측에서 레코드를 버리므로 빈 토픽에 대해 시험이 무의미하게 통과한다.

## 12. negative-space probes

**12.1 도달성 — 이 리프의 절반이 조립되지 않는다.** 스타터는 이렇게 만든다.

```java
return new dev.caskeleton.messaging.kafka.KafkaMessagingTransport("kafka", 1L, producer);
```

인자 셋짜리 생성자다. 그 생성자는 소비자 팩토리를 이렇게 채운다.

```java
spec -> { throw new MessagingCapabilityUnavailableException(
    "KAFKA_CONSUMER_NOT_CONFIGURED", "this Kafka transport was created without a consumer factory"); }
```

그리고 저장소 전체에서 `new KafkaConsumerRegistrar` 는 **테스트 5곳에만** 있다. 소비 경로 전체 — 폴 루프(621), 정착 큐, 오프셋 트래커, 재시도 스케줄러, 파티션 조정자, 배달 매퍼, 재시도 실행기 — 가 배포에 조립되지 않는다.

조립되는 것은 발행 경로다. 전송·발행 매퍼·헤더 매퍼·실패 분류기·`KafkaPosition`, 그리고 시작 검증기 하나.

이 사실이 §17.3·§17.4·§17.5 의 등급을 한 칸 낮춘다. 오늘의 사고가 아니라 소비를 배선하는 날의 사고다.

**12.2 참조가 0인 production 파일.** `KafkaBatchConsumerRegistrar` 146줄은 저장소 전체에서 자기 파일 밖의 참조가 없다 — production 도 테스트도 아니다. 배치 소비의 규칙(파티션을 넘지 않는 배치, `DESTINATION` 순서와의 비양립)을 정확하게 서술하고 아무도 부르지 않는다.

**12.3 대조군 — 능력 선언 방식.** 세 어댑터가 모두 능력을 상수로 둔다. Pulsar 는 전송과 검증기가 서로 다른 값을 답하고, NATS 는 프로파일과 무관하게 중복 제거를 참으로 둔다. Kafka 는 §17.2 의 형태다.

**12.4 테스트가 볼 수 없는 것.** 소비 경로의 세 결함이 전부 같은 이유로 시험에서 벗어난다.

| 결함 | 가리는 형태 |
|---|---|
| §17.3 천장 일시정지 후 재개 없음 | 결정적 시험은 천장에 닿은 **그 주기**까지만 단언한다(`anOrderedDestinationDispatchesOneRecordAtATime`). 실브로커 IT 는 전부 핸들러 풀이 `Runnable::run`(인라인)이라 천장에 닿지 않고, 전부 레코드 1건만 발행한다 |
| §17.4 재시도 헤더 오염 | `ReservedHeaderForgeryTest.aMalformedRetryAttemptIsQuarantined` 가 `attemptOf` 의 **던짐만** 단언한다. 소비자가 그 던짐을 어떻게 다루는지는 어떤 시험도 보지 않는다 |
| §17.5 벽시계 | `pollOnce(Instant)` 는 시계를 주입받는데 재시도 등록만 `Instant.now()` 를 읽는다. 지연 재개를 결정적으로 시험할 수 없다 |

**12.5 고쳐진 메서드와 증명된 메서드가 다르다.** §17.6.

## 16. 확인하지 못한 것

- 실제 브로커로 재조정 중 정착 거부를 재현하지 않았다. 인증 레인이 그 자리이고 컨테이너가 필요하다.
- §17.3 을 실행으로 재현하지 않았다. `consumer.resume(...)` 호출처가 둘(`applyDueResumes`·공개 `resume(scope)`)뿐이고 천장 경로가 `retries` 에 아무것도 등록하지 않는다는 것으로 판정했다.
- §17.4 를 실행으로 재현하지 않았다. `attemptOf` 가 `dispatch` 의 두 번째 `try` 안에 있고 그 `catch` 가 `requeueAfterFailure()` 라는 것, `MessagingConfigurationException` 이 `RuntimeException` 을 상속한다는 것으로 판정했다.
- Toxiproxy 인증 레인을 직접 돌리지 않았다. 코드와 그 레인이 기록하는 증거 형식만 읽었다.
- `gradle.lockfile` 은 읽지 않았다(`STRUCTURAL_ONLY`).

## 17. 손볼 것

### 17.1 P1 — 지원 문서가 `deduplicatedPublish` 를 지원으로 적고, 코드는 거짓이며, 그 차이가 정확히 코드가 경고한 피해다

```java
private static final MessagingCapabilities CAPABILITIES =
    new MessagingCapabilities(true, true, true, true, true, true, true, false, true, false, false, true);
//                                                                                      ^^^^^ deduplicatedPublish
```

코드의 판정이 옳고 그 근거가 javadoc 에 있다.

> "Producer idempotence deduplicates *sequence retries within one producer session*: the producer id
> is reassigned on restart, so the same logical message published again after a crash is a new
> sequence and the broker stores it twice."

`docs/messaging/support-matrix.md:55` 의 능력 표는 이 칸을 `O` 로 적는다.

그 차이가 무거운 이유는 이 플랫폼에서 이 플래그가 특별하기 때문이다. 능력 열둘 중 **부재가 예외를 만드는 유일한 플래그**다.

```java
// DefaultMessagePublisher:249-252
if (options.deduplication().isPresent()
    && !transport.capabilities(profile.name()).capabilities().deduplicatedPublish()) {
  throw new …("PUBLISH_DEDUPLICATION_UNSUPPORTED", …);
```

그래서 표를 읽고 중복 제거를 전제한 목적지를 설계한 팀은 실행 시점에 능력 예외를 만난다. 반대로 표를 읽고 "중복 제거가 있으니 모호를 그냥 재시도해도 된다" 고 결론지으면, 실제로는 중복이 저장된다.

`MessagingCapabilities` 의 클래스 javadoc 이 그 피해를 미리 적는다 — "a silently weakened guarantee is indistinguishable from a working one until the incident."

수정은 문서 쪽이다. 코드가 이미 옳다.

### 17.2 P2 — 브로커 트랜잭션을 무조건 참으로 선언하고, 그 조건을 검사하는 검증기는 시작 시 돌지 않는다

능력 상수의 아홉 번째가 `brokerTransaction = true` 다. 프로파일과 무관한 상수다.

그런데 Kafka 트랜잭션이 실제로 성립하려면 `KafkaTransactionProfileValidator` 가 요구하는 넷이 모두 참이어야 한다 — 트랜잭션 식별자 접두, 멱등 생산자, `acks=all`, 수동 커밋.

그 검증기는 스타터가 빈으로 만들지만 `StartupProfileValidation` 으로 감싸지 않는다.

```java
// KafkaMessagingAutoConfiguration
@Bean public KafkaProfileValidator kafkaProfileValidator() { … }
@Bean public StartupProfileValidation<KafkaBrokerProfile> kafkaProfileStartupValidation(…) { … }  // ← 감싼다
@Bean public KafkaTransactionProfileValidator kafkaTransactionProfileValidator() { … }            // ← 감싸지 않는다
```

즉 두 겹이 함께 비어 있다. 능력은 조건과 무관하게 참을 답하고, 조건을 검사할 검증기는 발행되기만 하고 주입되지 않는다.

`StartupProfileValidation` 의 javadoc 이 서술한 이전 결함이 정확히 그 형태다 — "the context published a validator per broker and validated nothing."

수정은 두 갈래를 함께 한다.

- 스타터에서 `kafkaProfileStartupValidation` 형태를 복사해 트랜잭션 검증기를 감싼다(스타터 SSOT §17.2 와 같은 수정).
- 능력을 프로파일에서 파생시킨다 — `enableIdempotence && "all".equals(acks) && 접두 존재`.

두 번째가 없으면 검증기가 돌더라도 능력 조회는 여전히 프로파일과 무관하게 답한다.

### 17.3 P2 — 천장에 닿아 일시정지된 파티션을 재개하는 경로가 없다

`pollOnce` 의 파티션 루프는 세 경우에 그 파티션을 멈춘다.

```java
if (!coordinator.tryAcquire(partition))  { seekBackTo = record.offset(); continue; }   // 천장
if (!shutdown.tryBeginWork())            { …                             continue; }   // 배수 시작
if (!dispatch(record, partition, now, epoch)) { …                        continue; }   // 풀 거부
…
if (seekBackTo >= 0) {
  consumer.pause(Set.of(partition));
  consumer.seek(partition, seekBackTo);
}
```

이 세 경로 중 어느 것도 `retries.pauseUntil(...)` 을 부르지 않는다. 그런데 폴 루프가 파티션을 재개하는 곳은 하나뿐이다.

```java
private void applyDueResumes(Instant now) {
  Map<TopicPartition, Long> due = retries.dueForResume(now);   // ← retries 에 등록된 것만
  due.forEach((partition, seekTo) -> { consumer.seek(...); coordinator.resume(...); consumer.resume(...); });
}
```

`retries` 에 항목을 넣는 곳은 `QueuedSettlement.enqueueRequeue` 하나이고, 그것은 핸들러 실패·타임아웃·명시적 requeue 경로다. 천장·배수·풀 거부 경로는 등록하지 않는다.

따라서 천장 때문에 멈춘 파티션은 **폴 루프가 스스로 재개하지 않는다.** 재개할 수 있는 것은 외부에서 부른 `resume(scope)` 이나 재조정뿐이다.

**도달 조건이 좁지 않다.** `maxInFlightPerOrderingUnit` 의 기본값은 1 이다(`DestinationSettings.Consumer`). 한 폴이 같은 파티션의 레코드를 둘 이상 돌려주는 순간 두 번째에서 `tryAcquire` 가 거짓이 되고, 그 파티션이 멈춘다. 그 뒤 작업자가 끝나 `coordinator.release` 로 슬롯이 비어도 `consumer` 는 여전히 일시정지 상태다.

**대조.** 같은 파일이 `coordinator.pause(...)` 와 `consumer.pause(...)` 를 구분해서 쓴다 — `applySettlements` 의 `PAUSE_AND_SEEK` 는 둘 다 부르고, 천장 경로는 `consumer` 쪽만 부른다. 그래서 조정자는 그 파티션을 멈춘 것으로 알지 못하고, 결과적으로 `tryAcquire` 는 계속 참을 답하는데 브로커에서 레코드가 오지 않는다.

**수정.** 천장 경로가 `retries.pauseUntil(partition, seekBackTo, Duration.ZERO, now)` 를 등록하면 다음 주기의 `applyDueResumes` 가 즉시 재개한다. 지연이 0 이므로 `dueForResume` 이 곧바로 돌려준다. 배수 경로는 재개하지 않는 것이 맞고, 풀 거부 경로는 천장과 같다.

**등급.** 소비 경로가 조립되지 않으므로(§12.1) P2. 배선하는 순간 P1 이다 — 파티션이 조용히 멈추고, 커밋 워터마크도 함께 멈추므로 소비자 랙만 늘어난다.

### 17.4 P2 — 오염된 재시도 헤더가 격리되지 않고 무한 pause-and-seek 을 만든다

`KafkaRetryMetadataMapper.attemptOf` 는 읽을 수 없는 `msg.retry.attempt` 에 대해 fail-closed 를 택하고, 그 이유를 정확하게 적는다.

```java
} catch (NumberFormatException malformed) {
  // Returning 1 for an unreadable header restarts the retry budget on every redelivery… and the
  // header is caller-influenced, which makes "unreadable" a way to defeat the cap rather than an
  // accident.
  throw new MessagingConfigurationException("RETRY_ATTEMPT_MALFORMED",
      "the retry attempt header is not a positive integer; the message is quarantined rather"
          + " than restarting its retry budget");
}
```

메시지가 "quarantined" 라고 말한다. 소비자는 그렇게 하지 않는다.

```java
try {
  envelope = deliveryMapper.toEnvelope(record);
} catch (RuntimeException undecodable) {
  if (quarantine.quarantine(record, undecodable)) { settlement.acknowledgeAfterQuarantine(); }
  else { settlement.requeueAfterFailure(); }
  … return;                                        // ← 격리 경로는 여기까지다
}
try {
  int attempt = retryMetadataMapper.attemptOf(envelope);     // ← 던지는 자리는 여기다
  …
} catch (ExecutionException | RuntimeException failure) {
  settlement.requeueAfterFailure();                          // ← pause-and-seek
}
```

격리 경로는 **디코딩 실패에만** 걸려 있다. `attemptOf` 는 디코딩이 끝난 뒤 두 번째 블록에서 던지고, `MessagingConfigurationException` 은 `MessagingException` 을 통해 `RuntimeException` 이므로 두 번째 `catch` 가 잡는다. 결과는 `requeueAfterFailure()` → `PAUSE_AND_SEEK` → 같은 오프셋 재읽기 → 같은 헤더 → 같은 예외다.

즉 fail-closed 가 막으려던 것(재시도 예산 무력화)보다 나쁜 것을 만든다 — 그 파티션이 영구히 그 레코드에서 멈춘다. 그리고 javadoc 이 지적한 대로 이 헤더는 호출자가 쓸 수 있는 값이므로, 숫자가 아닌 값 하나로 파티션 하나를 정지시킬 수 있다.

**테스트가 보지 못하는 이유.** `ReservedHeaderForgeryTest.aMalformedRetryAttemptIsQuarantined` 는 `attemptOf` 가 던지는 것만 단언한다. 이름은 "quarantined" 인데 격리를 확인하지 않는다.

**수정.** `attemptOf` 호출을 디코딩과 같은 블록으로 옮겨 격리 경로에 태우거나, 두 번째 `catch` 가 예외 종류를 나누게 한다 — `MessagingConfigurationException` 은 재시도로 회복되지 않는 종류이므로 격리 대상이고, 핸들러 실패는 재시도 대상이다.

### 17.5 P3 — 시계를 주입받는 클래스가 한 곳에서만 벽시계를 읽는다

`KafkaConsumerRegistrar` 의 설계 성질이 javadoc 에 적혀 있다.

> "`pollOnce(Instant)` is one full cycle and is public so the whole loop — commit ordering, pause,
> seek, rebalance — is testable against `MockConsumer` without threads or sleeps."

주기마다 `Instant now` 를 받아 `applyDueResumes(now)` 로 넘긴다. 그런데 그 짝인 등록 쪽은 이렇다.

```java
private SettlementResult enqueueRequeue(Duration delay) {
  retries.pauseUntil(partition, offset, delay, Instant.now());   // ← 주입된 시계가 아니다
```

이 리프에서 `Instant.now()` 를 읽는 유일한 자리다. 그리고 그 호출은 작업자 스레드에서 일어나므로 폴 스레드의 `now` 와 다른 순간이다.

결과는 두 가지다. 지연 재시도(`requeue(Duration)`)의 재개 시점을 고정 시계로 시험할 수 없고, 시험이 `Duration.ZERO` 밖의 지연을 다루지 못한다 — 실제로 어떤 시험도 다루지 않는다.

수정은 생성자에 `Supplier<Instant>` 를 하나 더 받는 것이다. 같은 저장소의 `MessagingShutdownLifecycle` 이 정확히 그 형태로 두 생성자를 둔다.

### 17.6 P3 — 결함으로 판정된 메서드가 남아 있고, 실브로커 증명이 그것 위에서 돈다

`KafkaTransactionalPublisher` 에 같은 일을 하는 메서드가 둘 있다.

```java
public <T> T inTransaction(delivery, profiles, inputOffsets, Supplier<T> body)   // begin → body → send → commit
public void sendInTransaction(delivery, profiles, inputOffsets)                   // begin → send → commit  (body 없음)
```

`inTransaction` 의 javadoc 이 둘째를 결함으로 지목한다.

> "The processor used to run the handler and only afterwards hand the delivery here — so
> `beginTransaction` happened *after* the handler had already finished… a handler that succeeded and
> a commit that then failed left the handler's work applied with its input offsets unsent."

`sendInTransaction` 은 public 이고 production 호출자가 없다. 호출하는 것은 시험 다섯 자리뿐이다 — 그리고 그 다섯이 실브로커 트랜잭션 증명 전부다(`KafkaTransactionIT`·`KafkaTransactionFencingIT`·`KafkaReadCommittedIT`).

고쳐진 `inTransaction` 을 시험하는 것은 `KafkaTransactionOrderingTest` 하나이고 `MockProducer` 다. 즉 실브로커에서 커밋·중단·펜싱이 증명된 것은 옛 모양이고, 새 모양은 목 위에서만 증명됐다.

기능적 차이는 크지 않다(`body` 가 비어 있으면 두 메서드는 같은 호출열을 만든다). 그래도 두 가지가 남는다 — 결함으로 판정된 순서를 만드는 public 진입점이 여전히 열려 있다는 것, 그리고 실브로커 증거가 production 경로가 아닌 것 위에 있다는 것.

수정은 ITs 를 `inTransaction(..., () -> null)` 로 옮기고 `sendInTransaction` 을 지우는 것이다.

### 확인된 설계(문제 아님)

- **모든 소비자 호출을 폴 스레드로 모으고, 그 이유를 클라이언트의 문서화된 비스레드안전성에서 끌어온 것.** 통독으로 실제 준수를 확인했다.
- **공개 제어 API 도 큐를 지나게 하고, 반환 단계가 다음 주기에 완료된다는 것을 정직하게 서술한 것.**
- **`close()` 만 큐를 지나지 않게 하고 그 예외에 근거를 붙인 것.**
- **연속 워터마크로만 커밋하고, 그 대가를 소비자 랙으로 받아들인 것.**
- **제출되지 않은 배달 등록을 되돌리는 경로.**
- **커밋 뒤에 지역 상태를 갱신하도록 순서를 고치고, 이전 순서가 만든 결함을 주석에 남긴 것.**
- **재조정에서 에폭을 먼저 지워 늦은 정착을 거부하는 것, 그리고 할당에서도 에폭을 파생시켜 수동 할당을 덮은 것.**
- **거부된 정착 수를 지표로 노출한 것.**
- **정착을 단일 종결(CAS)로 만든 것.**
- **파티션별 처리로 바꿔 전역 break 이 만들던 처리 간극을 없앤 것.**
- **알 수 없는 발행 실패의 기본값을 모호로 둔 것.**
- **격리 기본 구현이 `false` 를 답해 커밋을 막는 것** — 쓸 곳이 없으면 오프셋을 넘기지 않는다.
- **Kafka 트랜잭션이 데이터베이스 부수효과를 덮지 못한다는 것을 검증기가 거부로 표현한 것.**
- **재생 기본값을 격리된 임시 그룹으로 두고, 운영 그룹 재생에 승인을 요구한 것.**
- **오프셋 재설정의 승인 술어를 생성자 인자로 둔 것** — 승인 출처 없이 조립된 런타임은 물리적으로 재설정할 수 없다.
- **JAAS 값 이스케이프 순서(역슬래시 먼저)와 제어문자 거부.**
- **OAuth 를 절반만 설정하는 대신 거부한 것.**
- **예약 헤더 위조 거부를 "envelope 필드를 재진술하는 이름" 으로만 좁힌 것** — 재시도·사후처리 재발행이 그 가드에 걸리지 않는다.
- **인증 레인이 Docker 조건부 skip 을 쓰지 않는 것, 그리고 커버 못 하는 시나리오를 `knownGaps` 로 남긴 것.**
- **`pollOnce` 를 공개해 전체 주기를 스레드 없이 검증 가능하게 만든 것.**

---

## Source anchors

```
src/messaging/messaging-kafka/build.gradle
main/java/…/kafka/KafkaConsumerRegistrar.java:1-621   (§17.3 pollOnce:216-257 · dispatch:264-333 · applyDueResumes:375-383)
main/java/…/kafka/KafkaMessagingTransport.java:1-215  (능력 상수 62-64)
main/java/…/kafka/KafkaTransactionalPublisher.java:1-178  (§17.6 inTransaction:96-119 · sendInTransaction:144-178)
main/java/…/kafka/KafkaSecurityConfigurer.java:1-173
main/java/…/kafka/KafkaHeaderMapper.java:1-169
main/java/…/kafka/KafkaDeliveryMapper.java:1-167
main/java/…/kafka/KafkaBatchConsumerRegistrar.java:1-146  (§12.2 참조 0)
main/java/…/kafka/PartitionWorkCoordinator.java:1-128
main/java/…/kafka/KafkaRetryExecutor.java:1-126
main/java/…/kafka/ContiguousPartitionOffsetTracker.java:1-119
main/java/…/kafka/KafkaPublishFailureClassifier.java:1-116
main/java/…/kafka/KafkaPublishMapper.java:1-111
main/java/…/kafka/KafkaRetryMetadataMapper.java:1-93   (§17.4 attemptOf:725-748)
main/java/…/kafka/KafkaPartitionRetryScheduler.java:1-90
main/java/…/kafka/KafkaReplayCapability.java:1-82
main/java/…/kafka/KafkaRetryTopicPublisher.java:1-75
main/java/…/kafka/SpringKafkaTransactionalProcessor.java:1-68
main/java/…/kafka/KafkaOffsetResetExecutor.java:1-63
main/java/…/kafka/KafkaProfileValidator.java:1-60
main/java/…/kafka/KafkaReplayPlanner.java:1-58
main/java/…/kafka/KafkaBrokerProfile.java:1-56
main/java/…/kafka/{KafkaSettlementQueue:1-52, KafkaTransactionProfileValidator:1-52, KafkaSettlementCommand:1-50,
                   KafkaDeadLetterPublisher:1-49, PartitionOffsetTracker:1-48, KafkaTopologyInspector:1-45,
                   KafkaRetryOutcome:1-40, KafkaPosition:1-38, KafkaReplayPlan:1-36, KafkaQuarantinePublisher:1-31,
                   KafkaTransactionalProcessor:1-29, KafkaTransactionalDelivery:1-22, KafkaTransactionalOutput:1-21}
test/java/…/kafka/ 24파일 4,087줄 (KafkaBrokerCertificationIT:466 · KafkaConsumerRegistrarTest:440 · KafkaContractHarness:370 …)
messaging-spring-boot-starter/…/KafkaMessagingAutoConfiguration.java:100-125 (§12.1 발행 전용 조립 · §17.2)
messaging-runtime-core/…/DefaultMessagePublisher.java:249-252 (§17.1 능력 부재가 예외를 만드는 유일한 자리)
docs/messaging/support-matrix.md:55 (§17.1 능력 표 대조)
```