Kafka는 대규모 데이터 스트림을 처리하기 위한 분산형 메시징 플랫폼을 말한다.

단순한 메시지 큐가 아닌 분산 로그 시스템이기 때문에 내부 구성 요소들이 "고성능 / 장애 복구 / 확장성"을 중심으로 설계되어 있다.

🔹 Kafka 핵심 개념 요약

🔸 Producer (생산자)

• 메시지를 만들어서 Kafka에 전송하는 주체
• 예: 주문 서비스가 "주문 생성 이벤트"를 발행

🔸 Consumer (소비자)

• Kafka로부터 메시지를 읽어 처리하는 주체
• 예: 배송 서비스가 "주문 생성 이벤트"를 구독해서 배송 요청 처리

🔸 Broker (브로커)

• 메시지를 저장하고 전달하는 역할의 Kafka 서버 한 대
• 여러 대가 모여 Kafka Cluster를 구성
• 브로커 장애 시에도 데이터 유지 가능

🔸 Topic (토픽)

• 메시지가 구분되어 저장되는 논리적 분류 단위
• 예: order-created, payment-completed

🔸 Partition (파티션)

• 토픽을 쪼갠 물리적 단위
• 토픽을 여러 조각으로 나누어 병렬처리와 확장성 확보
• 각 파티션은 파티션 내부에서만 순서를 보장한다
• Partition 수만큼 Consumer 병렬 처리 가능
• 파티션 증가 = 처리량 증가

🔸 Offset (오프셋)

• 파티션 내 메시지의 순서 번호
• Consumer가 어디까지 읽었는지 추적하는 기준

🔹 Kafka 동작 구조

Producer  →  [Kafka Broker]  →  Consumer
               | Topic |
           ┌──────────────┐
           │ Partition 0  │ → 메시지 A, B, C ...
           │ Partition 1  │ → 메시지 D, E, F ...
           └──────────────┘

1. Producer가 특정 Topic으로 메시지를 보냄
2. Kafka Broker는 그 메시지를 해당 Topic의 Partition에 저장
3. Consumer는 자신이 구독한 Topic의 Partition에서 메시지를 Offset 순서대로 읽음

🔹 Kafka의 주요 특징

🔸 고성능 / 고처리량

• 초당 수십~수백만 건의 메시지 처리 가능

🔸 확장성

• 브로커와 파티션을 늘려 부하를 분산 가능

🔸 내구성 / 장애 복구

• 메시지를 디스크에 저장하고 복제하여 장애 시에도 데이터 유실 방지

🔸 실시간 처리

• 스트림 데이터를 실시간으로 처리
• 예: IoT 센서, 로그, 결제 이벤트 등

🔸 Pub / Sub 구조

• 한 Producer가 보낸 메시지를 여러 Consumer가 동시에 구독 가능

🔹 Kafka의 활용 예시

🔸 로그 수집 이벤트

• 각 서버의 로그를 Kafka에 모은 뒤 ELK(ElasticSearch, Logstash, Kibana)로 전송

🔸 주문 / 결제 이벤트 처리

• 주문, 결제, 배송 서비스를 비동기 이벤트 기반으로 연결

🔸 실시간 데이터 분석

• 클릭스트림, 트래픽 데이터, 금융 거래를 실시간 분석

🔸 IoT 데이터 스트리밍

• 수많은 센서 데이터 실시간 수집 및 처리

🔸 데이터 파이프라인 중간 버퍼

• Spark, Flink, Hadoop 등으로 데이터를 전달하는 중간 허브 역할

🔹 Kafka의 파티션 전략

Kafka에서 파티션 전략은 성능 / 순서 / 확장성 / 데이터 일관성을 동시에 결정하는 핵심 설계 요소이다. 그렇기 때문에 단순히 "많이 나누면 좋다"가 아니라, 비즈니스 특성과 소비 패턴을 고려해 설계해야 한다.

🔸 Key가 없는 경우

kafkaTemplate.send("order-topic", value);

• Kafka가 자동으로 Round Robin
• 메시지 균등 분산
• 순서 보장 없음 (전체 단위로)
• 예: 로그 수집, 클릭 이벤트, 순서 의미가 약한 데이터 등

🔸 Key 기반 파티셔닝

kafkaTemplate.send("order-topic", userId, value);

"이 단위 안에서는 순서를 반드시 보장해야 한다" → 그 단위를 key로 사용

🔸 key 기반 파티셔닝 예시

• 사용자별 순서 중요 → userId
• 주문별 순서 중요 → orderId
• 상품 재고 중요 → productId
• 전체 순서 중요 → Partition 1개

🔸 Hot Partition 문제

• 예: userId 기준으로 분산했는데 특정 user가 트래픽 80% 차지 → 결과:

    Partition 0 → 80% 부하
     나머지 → 한가함
  

  이를 Hot Partition이라고 한다.

• Hot Partition 문제에 대한 해결 전략으로 3개 정도 서술하자면,

  1. Key 세분화
  2. Composite Key 전략
  3. Custom Partitioner 사용

🔸 1. Key 세분화

userId + randomSuffix

단점: 순서 보장 깨질 수 있음

🔸 2. Composite Key 전략

userId + orderDate

시간 단위로 분산

🔸 3. Custom Partitioner 사용

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(...) {
        // 직접 분산 로직 작성
    }
}

특정 로직 기반 분산 가능

🔸 Partition 개수 설계 기준

파티션 수는 늘릴 수는 있지만 줄일 때는 위험(순서 보장 깨질 수 있음, 데이터 분산 재정렬)이 따르기 때문에 처음 설계가 매우 중요하다.

• 파티션 전략 설게 시 고려할 점:

  • 어디까지 순서를 보장해야 하는가?
  • 최대 트래픽은 얼마나 되는가?
  • 특정 key에 트래픽이 몰릴 가능성은 없는가?

• 목표 처리량을 기반으로 계산해서 설계할 때는:

  • 목표 TPS ÷ 파티션당 처리 가능 TPS = 최소 파티션 수

• Consumer 수 고려해서 설계할 때는

  • Consumer Group 내 Consumer 수 <= Partition 수
  • 향후 확장 고려해서 여유 두기

🔸 Partition 너무 많으면 생기는 문제

일반적으로 수십~수백 개는 흔함. 수천 개 이상은 신중해야 함

🔹 Kafka Outbox 패턴 적용

Kafka + Outbox 패턴 설계는 데이터 정합성과 확장성을 동시에 만족시키기 위한 핵심 설계이다. 특히 주문/결제/재고처럼 정합성이 중요한 도메인에서 거의 표준에 가깝다고 생각한다.

Outbox 패턴의 핵심은:

이벤트를 DB 트랜잭션 안에서 함께 저장하고 나중에 별도 프로세스가 Kafka로 발행하는 것에 있다.

• Outbox가 해결하는 것:
  • DB와 Kafka 간 원자성
  • Kafka 전송 실패 시 재시도 가능
  • 메시지 유실 방지
• 하지만, Kafka 레벨에서 중복은 발생 가능
• 따라서 Consumer는:
  • 멱등성 처리 필요
  • 이벤트 ID 기반 중복 제거

🔸 알아두면 좋은 실무에서 무난하고 안전한 설계 공식

1. Outbox 테이블에 event_id 포함
2. Kafka key = aggregate_id
3. Consumer는 event_id로 멱등 처리
4. Partition 수는 예상 TPS 기반 설계

[Order Service]
   ├─ Order 저장
   └─ Outbox 저장 (aggregate_id 포함)

[Outbox Publisher]
   └─ Kafka 전송 (key = aggregate_id)

[Kafka]
   └─ 같은 aggregate는 같은 Partition

[Consumer]
   └─ event_id 기반 멱등 처리

🔸 Kafka 장애 시 Outbox 재처리 전략

Kafka 장애 시 Outbox 재처리 전략의 핵심은:

1. 데이터 유실 없이 2. 중복은 허용하되 멱등으로 해결한다 3. 무한 재시도는 막는다 4. 락 전략을 명확히 한다

🔸 전체 장애 대응 흐름

DB TX 성공
   ↓
Outbox 저장
   ↓
Kafka 장애 발생
   ↓
Outbox 재시도
   ↓
Kafka 복구
   ↓
정상 전송
   ↓
Consumer 멱등 처리

🔸 장애 시나리오별 분석

🔸 시나리오 1. Kafka 브로커 일시적 다운

[상황]

• DB 저장 성공
• Outbox 저장 성공
• Kafka 전송 시도 → 실패

[해결 전략]

• Outbox row는 status = READY 상태 유지 → Publisher가 재시도

🔸 시나리오 2. Kafka 전송 성공했지만 응답 받기 전 장애

[상황]

• Kafka에 실제로 메시지는 저장됨
• 네트워크 문제로 응답 못 받음
• 애플리케이션은 실패로 인식
• 재전송 발생 → 중복 이벤트 발생 가능

[해결 전략]

• Consumer는 반드시 멱등 처리

🔸 시나리오 3. Outbox Publisher 장애

[상황]

• 애플리케이션 정상
• Outbox 쌓임
• Publisher 프로세스 다운

[해결 전략]

• 재기동 후 미처리 row 재전송 필요

🔸 Outbox 재처리 흐름

1. READY 조회
2. PROCESSING 변경 (락 확보)
3. Kafka 전송 시도
4. 성공 → SENT
5. 실패 → READY (retry_count++)

🔸 outbox의 동시성 제어 전략

멀티 인스턴스 환경에서는 같은 row를 여러 Publisher가 가져가면 안 된다.

해결방법 1. DB Row Lock (Pessimistic Lock)

SELECT *
FROM outbox
WHERE status = 'READY'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 100;

동시에 여러 인스턴스가 실행해도 중복 처리 방지

• 조회와 동시에 row-level lock 획득
• 다른 트랜잭션은 해당 row 접근 불가
• SKIP LOCKED → 이미 잡힌 row는 건너뜀

장점

• 구현이 직관적
• 멀티 인스턴스 환경에서 안전
• 충돌 시 자동 회피

단점

• DB 락 경합 가능
• 처리 시간이 길면 lock 유지 시간 증가
• 트랜잭션 관리 신중해야 함

해결방법 2. 상태 전이 기반 Lock (Optimistic 방식)

UPDATE outbox
SET status = 'PROCESSING'
WHERE id = ?
AND status = 'READY';

update count == 1 이면 성공적으로 점유

• READY row 조회
• UPDATE로 상태 변경 시도
• 성공한 인스턴스만 처리

장점

• DB lock 유지 시간 짧음
• 대량 처리에 유리
• 확장성 좋음

단점

• 충돌 시 재시도 로직 필요
• 상태 꼬임 가능 (예: 장애 중단)

🔹 RabbitMQ와 비교

Kafka는 단순한 메시지 큐가 아니라 분산 로그 시스템이다. 그래서 내부 구성요소들이 “고성능 · 장애복구 · 확장성”을 중심으로 설계되어 있다.

🔹 Broker (브로커)

🔸 정의

• Kafka 서버 한 대
• 메시지를 저장하고 전달하는 역할
• 여러 Broker가 모여 Kafka Cluster를 구성

🔸 Broker의 핵심 책임

📌 중요 포인트

• Kafka는 메모리 큐가 아니라 디스크 로그
• Broker 장애 시에도 데이터 유지 가능

🔹 Topic & Partition

🔸 Topic

• 메시지의 논리적 분류 단위
• 예: order-created, payment-completed

🔸 Partition

• Topic을 쪼갠 물리적 단위
• 병렬 처리와 확장성을 담당

🔸 핵심 특징

📌 실무 핵심

• Kafka에서 성능 = Partition 수
• 단, Partition 늘리면 순서 보장 범위는 더 좁아짐

🔹 Replica & ISR (복제와 장애 복구)

🔸 Replica

• Partition의 복사본
• 여러 Broker에 분산 저장

🔸 Leader / Follower

🔸 ISR (In-Sync Replicas)

• Leader와 동기화 상태인 Replica 집합
• ISR에 속한 Replica만 Leader 승격 가능

📌 왜 중요한가?

• Leader Broker 장애 → ISR 중 하나가 즉시 Leader 승격
• 데이터 유실 없이 서비스 지속 가능

🔹 Controller (클러스터의 두뇌)

🔸 정의

• Kafka Cluster에서 단 하나만 존재
• Broker 중 하나가 Controller 역할 수행

🔸 Controller의 역할

📌 Controller 장애 시

• 다른 Broker가 자동으로 Controller 승격
• 서비스 중단 없음

🔹 Zookeeper vs KRaft (RAFT 기반)

🔸 과거: Zookeeper 기반

Zookeeper가 담당했던 것

• Broker 등록
• Controller 선출
• Topic / Partition 메타데이터 저장

❌ 문제점

• 운영 복잡도 ↑
• Kafka + Zookeeper 이중 관리
• 확장성 한계

🔸 현재/미래: KRaft (Kafka Raft)

Kafka 2.8+부터 도입된 Zookeeper 제거 아키텍처

📌 실무 기준

• 신규 Kafka 클러스터 → KRaft 권장
• 기존 운영 중 → 점진적 마이그레이션 고려

🔹 Producer → Broker 내부 흐름

1. Producer가 메시지 전송
2. Partition Leader가 메시지 저장
3. Follower Replica들이 복제
4. acks 설정에 따라 Producer 응답

🔸 acks 옵션

🔹 Consumer Group 내부 동작

🔸 Consumer Group

• 같은 Group ID를 가진 Consumer 묶음
• Partition은 Group 내에서 1 Consumer만 처리

🔸 Rebalance

• Consumer 추가/제거 시
• Partition 재분배 발생

📌 주의

• Rebalance 동안 메시지 소비 일시 중단
• Consumer 설계 시 처리 시간 최소화 중요

🔹 전체 구조 요약

[Producer]
     ↓
[Partition Leader (Broker)]
     ↓ (Replica)
[Follower Broker]
     ↓
[Consumer Group]

🔹 실무에서 꼭 기억할 포인트

1. Kafka = 분산 로그 시스템
2. 성능 튜닝의 핵심은 Partition 설계
3. 장애 대응은 Replica + ISR
4. 최신 Kafka는 Zookeeper 제거(KRaft) 방향
5. 이벤트 기반 아키텍처의 핵심 인프라

(DB 관점)

🔹 낙관적 락 (Optimistic Lock)

🔸 개념

• 충돌이 드물다고 가정
• 데이터를 읽을 때는 락을 걸지 않고, 업데이트 시점에 변경 여부를 검증
• 주로 버전(version) 컬럼 또는 타임스탬프로 검증

🔸 동작 방식

1. 트랜잭션 A/B가 같은 row를 읽음
2. A가 먼저 수정 → version 증가
3. B가 수정 시도 → version 불일치 → 업데이트 실패

🔸 예시 (JPA)

@Version
private Long version;

🔸 장점

• 락 대기 없음 → 성능 우수
• 데드락 위험 거의 없음
• 읽기 중심 시스템에 적합

🔸 단점

• 충돌 시 재시도 로직 필요
• 충돌 빈도가 높으면 오히려 비용 증가

🔸 적합한 상황

• 트래픽은 많지만 동시 수정은 드문 경우
• 조회 위주의 서비스
• REST API, 이벤트 기반 처리, MSA 환경

🔹 비관적 락 (Pessimistic Lock)

🔸 개념

• 충돌이 자주 발생한다고 가정
• 데이터를 읽는 순간부터 락을 선점
• 다른 트랜잭션은 대기 또는 실패

🔸 동작 방식

1. 트랜잭션 A가 row 조회 + 락 획득
2. B는 해당 row 접근 시 대기(block) 또는 실패
3. A 커밋/롤백 후 락 해제

🔸 예시 (JPA)

@Lock(LockModeType.PESSIMISTIC_WRITE)
@Query("select o from Order o where o.id = :id")
Order findForUpdate(Long id);

🔸 장점

• 데이터 정합성 강력 보장
• 충돌 처리 로직 단순

🔸 단점

• 락 대기 → 성능 저하
• 데드락 가능성
• 트랜잭션 길어지면 시스템 병목

🔸 적합한 상황

• 동시 수정이 빈번
• 금액, 재고, 좌석, 포인트 등 절대 정확해야 하는 데이터
• 짧은 트랜잭션이 보장될 때

🔹 핵심 비교 표

🔹 실무 선택 기준 (중요)

🔸 낙관적 락을 선택해야 할 때

• 평균적으로 동시 수정 확률 < 1%
• API 요청이 많고 응답 지연이 민감
• 재시도가 허용되는 도메인

🔸 비관적 락을 선택해야 할 때

• 절대 중복/오류가 발생하면 안 됨
• 충돌이 빈번하고 재시도가 비용 큼
• 트랜잭션이 매우 짧음

🔹 실무에서 자주 쓰는 패턴

• 기본은 낙관적 락

• 정말 중요한 구간만 비관적 락으로 국소 적용

• MSA 환경에서는 DB 락보다

  • Redis 분산 락
  • 메시지 큐 직렬화
  • Saga / Outbox 패턴 을 함께 고려

🔸 JpaOrderOutboxEventRepository

package chill_logistics.order_server.infrastructure.repository;

import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.entity.OrderOutboxStatus;
import jakarta.persistence.LockModeType;
import java.util.List;
import java.util.UUID;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Lock;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;

public interface JpaOrderOutboxEventRepository extends JpaRepository<OrderOutboxEvent, UUID> {

    /* 처리 대상으로 가져와서 락 거는 목적 용도 (PENDING + Lock + 오래된 순) */
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("SELECT e FROM OrderOutboxEvent e " +
           "WHERE e.orderOutboxStatus = :status " +
           "ORDER BY e.createdAt ASC")
    List<OrderOutboxEvent> findPendingEventsForUpdate(
        @Param("status") OrderOutboxStatus status,
        Pageable pageable);

    /* 검색/관리용 (status/orderId 조건 검색) */
    @Query("SELECT e FROM OrderOutboxEvent e " +
           "WHERE (:status IS NULL OR e.orderOutboxStatus = :status) " +
           "AND (:orderId IS NULL OR e.orderId = :orderId) " +
           "ORDER BY e.createdAt ASC")
    List<OrderOutboxEvent> findOutboxEvents(
        @Param("status") OrderOutboxStatus status,
        @Param("orderId") UUID orderId,
        Pageable pageable);

    /* FAILED 모니터링/조회용 */
    @Query("SELECT e FROM OrderOutboxEvent e " +
           "WHERE e.orderOutboxStatus = 'FAILED' " +
           "ORDER BY e.lastRetryAt DESC, e.createdAt DESC")
    List<OrderOutboxEvent> findFailedEvents(Pageable pageable);
}

🔸 OrderOutboxEventRepositoryAdapter

package chill_logistics.order_server.infrastructure.repository;

import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.entity.OrderOutboxStatus;
import chill_logistics.order_server.domain.repository.OrderOutboxEventRepository;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.springframework.data.domain.PageRequest;

@RequiredArgsConstructor
public class OrderOutboxEventRepositoryAdapter implements OrderOutboxEventRepository {

    private final JpaOrderOutboxEventRepository jpaRepository;

    @Override
    public OrderOutboxEvent save(OrderOutboxEvent event) {
        return jpaRepository.save(event);
    }

    @Override
    public Optional<OrderOutboxEvent> findById(UUID id) {
        return jpaRepository.findById(id);
    }

    @Override
    public List<OrderOutboxEvent> findPendingEvents(OrderOutboxStatus status, int batchSize) {
        return jpaRepository.findPendingEventsForUpdate(status, PageRequest.of(0, batchSize));
    }

    @Override
    public List<OrderOutboxEvent> findOutboxEvents(OrderOutboxStatus status, UUID orderId, int page, int size) {
        return jpaRepository.findOutboxEvents(status, orderId, PageRequest.of(page, size));
    }

    @Override
    public List<OrderOutboxEvent> findFailedEvents(int page, int size) {
        return jpaRepository.findFailedEvents(PageRequest.of(page, size));
    }
}

🔸 OrderOutboxEventRepository

package chill_logistics.order_server.domain.repository;

import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.entity.OrderOutboxStatus;
import java.util.List;
import java.util.Optional;
import java.util.UUID;

public interface OrderOutboxEventRepository {

    OrderOutboxEvent save(OrderOutboxEvent orderOutboxEvent);

    Optional<OrderOutboxEvent> findById(UUID orderOutboxEventId);

    List<OrderOutboxEvent> findPendingEvents(OrderOutboxStatus status, int batchSize);

    List<OrderOutboxEvent> findOutboxEvents(OrderOutboxStatus status, UUID orderId, int page, int size);

    List<OrderOutboxEvent> findFailedEvents(int page, int size);
}

🔸 OrderOutboxEvent

package chill_logistics.order_server.domain.entity;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.EntityListeners;
import jakarta.persistence.EnumType;
import jakarta.persistence.Enumerated;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.UUID;
import lib.entity.BaseEntity;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.NoArgsConstructor;
import org.hibernate.annotations.GenericGenerator;
import org.springframework.data.jpa.domain.support.AuditingEntityListener;

@Getter
@Entity
@Table(name = "p_order_outbox_event")
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@EntityListeners(AuditingEntityListener.class)
public class OrderOutboxEvent extends BaseEntity {

    @Id
    @GeneratedValue(generator = "uuidv7")
    @GenericGenerator(
        name = "uuidv7",
        strategy = "lib.id.UUIDv7Generator"
    )
    @Column(name = "id", columnDefinition = "BINARY(16)")
    private UUID id;

    @Column(name = "order_id", nullable = false, columnDefinition = "BINARY(16)")
    private UUID orderId;

    @Column(name = "event_type", nullable = false, length = 100)
    private String eventType;

    @Column(name = "payload", nullable = false, columnDefinition = "TEXT")
    private String payload;

    @Enumerated(EnumType.STRING)
    @Column(name = "order_outbox_status", nullable = false, length = 30)
    private OrderOutboxStatus orderOutboxStatus;

    @Column(name = "retry_count", nullable = false)
    private int retryCount;

    @Column(name = "published_at")
    private LocalDateTime publishedAt;

    @Column(name = "last_retry_at")
    private LocalDateTime lastRetryAt;

    private static final int MAX_RETRY_COUNT = 3;
    private static final long MAX_BACKOFF_MILLIS = 100L;

    private OrderOutboxEvent(UUID orderId, String eventType, String payload) {
        this.orderId = orderId;
        this.eventType = eventType;
        this.payload = payload;
        this.orderOutboxStatus = OrderOutboxStatus.PENDING;
        this.retryCount = 0;
    }

    public static OrderOutboxEvent create(UUID orderId, String eventType, String payload) {
        return new OrderOutboxEvent(orderId, eventType, payload);
    }

    /* 이미 처리 되었는지 */
    public boolean alreadyHandled() {
        return this.orderOutboxStatus == OrderOutboxStatus.PUBLISHED
            || this.orderOutboxStatus == OrderOutboxStatus.FAILED;
    }

    /* 재시도 해도 되는지 */
    public boolean canRetry() {
        return this.orderOutboxStatus == OrderOutboxStatus.PENDING
            && this.retryCount < MAX_RETRY_COUNT;
    }

    /* 지금 재시도 할 준비 되었는지 */
    public boolean readyToRetry() {

        if (!canRetry()) {
            return false;
        }

        // 첫 시도는 즉시 성공
        if (this.lastRetryAt == null) {
            return true;
        }

        long timeSinceLastRetryMillis =
            Duration.between(this.lastRetryAt, LocalDateTime.now()).toMillis();

        return timeSinceLastRetryMillis >= MAX_BACKOFF_MILLIS;
    }

    /* 성공 확정 */
    public void markPublished() {
        this.orderOutboxStatus = OrderOutboxStatus.PUBLISHED;
        this.publishedAt = LocalDateTime.now();
    }

    /* 실패 처리 */
    public void markFailed() {
        this.retryCount++;
        this.lastRetryAt = LocalDateTime.now();

        // 최대 초과 시 FAILED
        if (this.retryCount >= MAX_RETRY_COUNT) {
            this.orderOutboxStatus = OrderOutboxStatus.FAILED;
        }
    }
}

🔸 KafkaProducerConfig

package chill_logistics.order_server.infrastructure.config;

import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> outboxProducerFactory() {

        Map<String, Object> config = new HashMap<>();

        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // Outbox 패턴에서 권장되는 설정
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, String> outboxKafkaTemplate() {
        return new KafkaTemplate<>(outboxProducerFactory());
    }
}

🔸 OrderOutboxKafkaProducer

package chill_logistics.order_server.infrastructure.kafka.outbox;

import chill_logistics.order_server.domain.event.OutboxEventProducer;
import chill_logistics.order_server.infrastructure.kafka.KafkaPassportProducerSupport;
import chill_logistics.order_server.lib.error.ErrorCode;
import java.util.concurrent.CompletableFuture;
import lib.passport.PassportIssuer;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderOutboxKafkaProducer implements OutboxEventProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final PassportIssuer passportIssuer;

    @Value("${app.kafka.topic.order-after-create}")
    private String orderAfterCreateTopic;

    @Value("${app.kafka.topic.order-canceled}")
    private String orderCanceledTopic;

    @Override
    public CompletableFuture<SendResult<String, String>> publish(String eventType, String key, String payload) {

        String topic = resolveTopic(eventType);

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, payload);

        // ✅ passport/user/trace 헤더 삽입
        KafkaPassportProducerSupport.writeHeaders(record.headers(), passportIssuer);

        return kafkaTemplate.send(record);
    }

    private String resolveTopic(String eventType) {

        return switch (eventType) {
            case "OrderAfterCreateV1" -> orderAfterCreateTopic;
            case "OrderCanceledV1" -> orderCanceledTopic;
            default -> throw new BusinessException(ErrorCode.UNKNOWN_EVENT_TYPE);
        };
    }
}

🔸 OrderOutboxProcessor

package chill_logistics.order_server.infrastructure.kafka.outbox;

import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.entity.OrderOutboxStatus;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.PageRequest;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderOutboxProcessor {

    private static final int BATCH_SIZE = 200;                 // 한 번의 process 에 처리할 이벤트 건 수
    private static final long FIXED_DELAY_MS = 500;            // Processor 스케줄러 실행 주기

    private final OrderOutboxEventRepository outboxEventRepository;
    private final OutboxEventTransactionManager transactionManager;

    @Scheduled(fixedDelay = FIXED_DELAY_MS)
    public void publishPendingEvents() {

        // 상태가 PENDING인 이벤트만 조회 & createdAt 기준 오래된 이벤트부터 처리
        List<OrderOutboxEvent> eventList = outboxEventRepository.findPendingEvents(
            OrderOutboxStatus.PENDING,
            PageRequest.of(0, BATCH_SIZE)
        );

        if (eventList.isEmpty()) {
            return;
        }

        int processed = 0;
        int skipped = 0;

        for (OrderOutboxEvent event : eventList) {

            // 이미 처리되었거나, 재시도 조건 충족 안 되면 스킵
            if (event.alreadyHandled() || !event.readyToRetry()) {
                skipped++;
                continue;
            }

            // 이벤트 1건 → REQUIRES_NEW 트랜잭션에서 처리
            transactionManager.publishEvent(event.getId());
            processed++;
        }

        log.info("[OUTBOX 처리 완료] 조회건수={} 처리건수={} 스킵건수={}",
            eventList.size(), processed, skipped);
    }
}

🔸 OutboxEventTransactionManager

package chill_logistics.order_server.infrastructure.kafka.outbox;

import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.event.OutboxEventProducer;
import chill_logistics.order_server.lib.error.ErrorCode;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class OutboxEventTransactionManager {

    private static final long SEND_CONFIRM_TIMEOUT_MS = 3000;  // Kafka 전송 요청 보낸 뒤, 성공 응답 기다리는 최대 시간

    private final OrderOutboxEventRepository outboxEventRepository;
    private final OutboxEventProducer outboxEventProducer;

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void publishEvent(UUID outboxId) {

        OrderOutboxEvent event = outboxEventRepository.findById(outboxId)
            .orElseThrow(() -> new BusinessException(ErrorCode.ORDER_OUTBOX_EVENT_NOT_FOUND));

        // 이미 처리되었거나, 재시도 조건 충족 안 되면 종료
        if (event.alreadyHandled() || !event.readyToRetry()) {
            return;
        }

        String key = event.getOrderId().toString();

        try {
            // Kafka 전송 + 성공 확정 대기
            outboxEventProducer
                .publish(event.getEventType(), key, event.getPayload())
                .get(SEND_CONFIRM_TIMEOUT_MS, TimeUnit.MILLISECONDS);

            // 전송 성공 처리
            event.markPublished();

            log.info("[OUTBOX 이벤트 발행 성공] outboxId={} eventType={} orderId={}",
                event.getId(), event.getEventType(), event.getOrderId());

        } catch (Exception ex) {
            // 전송 실패 처리 (retryCount 증가)
            event.markFailed();

            log.warn("[OUTBOX 이벤트 발행 실패] outboxId={} 재시도 횟수={}",
                event.getId(), event.getRetryCount(), ex);
        }
    }
}

🔸 OrderCommandService

package chill_logistics.order_server.application.service;

import chill_logistics.order_server.application.dto.command.CreateOrderCommandV1;
import chill_logistics.order_server.application.dto.command.CreateOrderResultV1;
import chill_logistics.order_server.application.dto.command.FirmInfoV1;
import chill_logistics.order_server.application.dto.command.FirmResultV1;
import chill_logistics.order_server.application.dto.command.OrderAfterCreateV1;
import chill_logistics.order_server.application.dto.command.OrderCanceledV1;
import chill_logistics.order_server.application.dto.command.OrderProductInfoV1;
import chill_logistics.order_server.application.dto.command.ProductResultV1;
import chill_logistics.order_server.application.dto.command.ReceiverInfoV1;
import chill_logistics.order_server.application.dto.command.SupplierInfoV1;
import chill_logistics.order_server.application.dto.command.UpdateOrderStatusCommandV1;
import chill_logistics.order_server.domain.entity.Order;
import chill_logistics.order_server.domain.entity.OrderOutboxEvent;
import chill_logistics.order_server.domain.entity.OrderProduct;
import chill_logistics.order_server.domain.entity.OrderQuery;
import chill_logistics.order_server.domain.entity.OrderStatus;
import chill_logistics.order_server.domain.port.FirmPort;
import chill_logistics.order_server.domain.port.HubPort;
import chill_logistics.order_server.domain.port.ProductPort;
import chill_logistics.order_server.domain.repository.OrderQueryRepository;
import chill_logistics.order_server.domain.repository.OrderRepository;
import chill_logistics.order_server.lib.error.ErrorCode;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import lib.entity.Role;
import lib.util.SecurityUtils;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderCommandService {

    private static final String EVENT_ORDER_AFTER_CREATE = "OrderAfterCreateV1";
    private static final String EVENT_ORDER_CANCELED = "OrderCanceledV1";

    private final OrderRepository orderRepository;
    private final OrderQueryRepository orderQueryRepository;
    private final ProductPort productPort;
    private final HubPort hubPort;
    private final FirmPort firmPort;
    private final OrderOutboxEventRepository outboxEventRepository;
    private final ObjectMapper objectMapper;

    private Order readOrderOrThrow(UUID orderId) {
        return orderRepository.findById(orderId)
                .orElseThrow(() -> new BusinessException(ErrorCode.ORDER_NOT_FOUND));
    }

    // Outbox 이벤트 적재
    private void saveOutboxEvent(UUID orderId, String eventType, Object message) {

        final String payload;
        try {
            payload = objectMapper.writeValueAsString(message);

        } catch (JsonProcessingException e) {
            // Outbox payload 직렬화 실패는 주문 트랜잭션 자체를 실패시키는 게 일반적으로 안전
            log.error("[OUTBOX payload 직렬화 실패] orderId={} eventType={}", orderId, eventType, e);

            throw new BusinessException(ErrorCode.OUTBOX_PAYLOAD_SERIALIZATION_FAILED);
        }

        OrderOutboxEvent outboxEvent = OrderOutboxEvent.create(orderId, eventType, payload);

        outboxEventRepository.save(outboxEvent);

        log.info("[OUTBOX 이벤트 적재 완료] orderId={} eventType={} outboxId={}",
            orderId, eventType, outboxEvent.getId());
    }

    @Transactional
    public CreateOrderResultV1 createOrder(CreateOrderCommandV1 command) {

        // 업체 조회
        FirmResultV1 supplierResult = firmPort.readFirmById(command.supplierFirmId(), "SUPPLIER");
        FirmResultV1 receiverResult = firmPort.readFirmById(command.receiverFirmId(), "RECEIVER");

        // 주문 상품 체크 및 재고 감소
        List<OrderProductInfoV1> orderProductInfoList =
                command.productList()
                        .stream()
                        .map(p -> {
                            // 상품 조회
                            ProductResultV1 product = productPort.readProductById(p.productId());

                            // 공급 업체 소속 상품인지 체크
                            if (!product.firmId().equals(command.supplierFirmId())) {
                                throw new BusinessException(ErrorCode.PRODUCT_NOT_FROM_FIRM);
                            }

                            // 상품 재고 체크
                            if (product.stockQuantity() < p.quantity()) {
                                throw new BusinessException(ErrorCode.OUT_OF_STOCK);
                            }

                            // 상품 재고 감소
                            productPort.decreaseStock(p.productId(), p.quantity());

                            return new OrderProductInfoV1(
                                    p.productId(),
                                    product.name(),
                                    product.price(),
                                    p.quantity()
                            );
                        })
                        .toList();

        // 주문 생성
        Order order = Order.create(
                command.supplierFirmId(),
                command.receiverFirmId(),
                command.requestNote(),
                orderProductInfoList
        );

        Order createOrder = orderRepository.save(order);

        // 주문 생성 시 주문 읽기 생성
        // TODO: 추후 주문 읽기 전략 수정예정 (임시: 대표 상품)
        OrderQuery orderQuery = OrderQuery.create(
                createOrder,
                SupplierInfoV1.from(supplierResult),
                ReceiverInfoV1.from(receiverResult)
        );

        orderQueryRepository.save(orderQuery);

        // Kafka 메시지 생성 (즉시 발행하지 않음)
        OrderAfterCreateV1 message = new OrderAfterCreateV1(
                createOrder.getId(),
                supplierResult.hubId(),
                receiverResult.hubId(),
                createOrder.getReceiverFirmId(),
                receiverResult.firmFullAddress(),
                receiverResult.firmOwnerName(),
                createOrder.getRequestNote(),
                // TODO: 추후 주문 읽기 전략 수정예정 (임시: 대표 상품)
                createOrder.getOrderProductList().get(0).getProductName(),
                createOrder.getOrderProductList().get(0).getQuantity(),
                createOrder.getCreatedAt()
        );

        // Outbox 적재
        saveOutboxEvent(createOrder.getId(), EVENT_ORDER_AFTER_CREATE, message);

        return CreateOrderResultV1.from(
                createOrder,
                FirmInfoV1.from(supplierResult),
                FirmInfoV1.from(receiverResult)
        );
    }

    @Transactional
    public void updateOrderStatus(UUID id, UpdateOrderStatusCommandV1 command) {

        // 주문 조회
        Order order = readOrderOrThrow(id);

        // 담당 허브 소속 주문인지 체크
        if (SecurityUtils.hasRole(Role.HUB_MANAGER)) {
            List<UUID> managingHubId = hubPort.readHubId(SecurityUtils.getCurrentUserId());
            UUID receiverHubId = firmPort.readHubId(order.getReceiverFirmId());

            if (!managingHubId.contains(receiverHubId)) {
                throw new BusinessException(ErrorCode.ORDER_NOT_IN_MANAGING_HUB);
            }
        }

        // 상태 변경
        order.updateStatus(command.status());

        // TODO: 주문 읽기 업데이트 (OrderStatus)
    }

    @Transactional
    public void deleteOrder(UUID id) {

        // 주문 조회
        Order order = readOrderOrThrow(id);

        // 담당 허브 소속 주문인지 체크
        if (SecurityUtils.hasRole(Role.HUB_MANAGER)) {
            List<UUID> managingHubId = hubPort.readHubId(SecurityUtils.getCurrentUserId());
            UUID receiverHubId = firmPort.readHubId(order.getReceiverFirmId());

            if (!managingHubId.contains(receiverHubId)) {
                throw new BusinessException(ErrorCode.ORDER_NOT_IN_MANAGING_HUB);
            }
        }

        // 본인 주문인지 체크
        if (SecurityUtils.hasRole(Role.FIRM_MANAGER)) {
            UUID currentUserId = SecurityUtils.getCurrentUserId();

            if (!currentUserId.equals(order.getCreatedBy())) {
                throw new BusinessException(ErrorCode.ORDER_NOT_CREATED_BY_USER);
            }
        }

        order.updateStatus(OrderStatus.CANCELED);
        order.delete(SecurityUtils.getCurrentUserId());

        // 재고 복원
        for (OrderProduct p : order.getOrderProductList()) {
            productPort.recoverStock(p.getProductId(), p.getQuantity());
        }

        // Kafka 메시지 생성 (즉시 발행하지 않음)
        OrderCanceledV1 message = new OrderCanceledV1(
            order.getId(),
            order.getOrderStatus(),  // CANCELED
            LocalDateTime.now()
        );

        saveOutboxEvent(order.getId(), EVENT_ORDER_CANCELED, message);

        // TODO: 주문 읽기 업데이트 (OrderStatus, delete)
    }
}

import com.klp.common.BaseEntity;
import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.EnumType;
import jakarta.persistence.Enumerated;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import java.time.LocalDateTime;
import java.util.UUID;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.NoArgsConstructor;

@Entity
@Table(name = "p_order_outbox_events")
@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class OrderOutboxEvent extends BaseEntity {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false)
    private UUID orderId;

    @Column(nullable = false)
    private String eventType;

    @Column(columnDefinition = "TEXT", nullable = false)
    private String payload;

    @Enumerated(EnumType.STRING)
    @Column(nullable = false)
    private OrderOutboxStatus status;

    @Column(nullable = false)
    private Integer retryCount = 0;

    private LocalDateTime publishedAt;

    private LocalDateTime lastRetryAt;

    public static OrderOutboxEvent create(UUID orderId,
        String eventType, String payload) {
        OrderOutboxEvent event = new OrderOutboxEvent();
        event.orderId = orderId;
        event.eventType = eventType;
        event.payload = payload;
        event.status = OrderOutboxStatus.PENDING;
        return event;
    }

    private static final int MAX_RETRY_COUNT = 20;           // 최대 20회
    private static final long INITIAL_BACKOFF_MILLIS = 1000L;  // 초기 1초
    private static final long MAX_BACKOFF_MILLIS = 300000L;    // 최대 5분

    public void markAsPublishing() {
        this.status = OrderOutboxStatus.PUBLISHING;
        this.lastRetryAt = LocalDateTime.now();
        this.retryCount++;
    }

    public void markAsPublished() {
        this.status = OrderOutboxStatus.PUBLISHED;
        this.publishedAt = LocalDateTime.now();
    }

    public void markAsFailed() {
        this.retryCount++;
        this.lastRetryAt = LocalDateTime.now();

        // 20회 초과 시 FAILED 처리
        if (this.retryCount >= MAX_RETRY_COUNT) {
            this.status = OrderOutboxStatus.FAILED;
        }
    }

    public boolean canRetry() {
        return this.status != OrderOutboxStatus.PUBLISHED
            && this.status != OrderOutboxStatus.FAILED;
    }

    /**
     * AWS/Google Cloud 권장 전략 1초 → 2초 → 4초 → 8초 → ... → 최대 5분
     */
    // 1초 2초 ... 최대 5분 -> AWS, GOOGLE 방식
    public long getBackoffMillis() {
        // Exponential: 1초 * 2^(retryCount-1)
        long exponentialBackoff = INITIAL_BACKOFF_MILLIS * (1L << (this.retryCount - 1));

        // 최대값 제한
        long cappedBackoff = Math.min(exponentialBackoff, MAX_BACKOFF_MILLIS);

        // Jitter 추가 (0~20% 랜덤 변동)
        double jitterFactor = 0.8 + (Math.random() * 0.4); // 0.8 ~ 1.2
        return (long) (cappedBackoff * jitterFactor);
    }

    public boolean shouldRetryNow() {
        if (!canRetry()) {
            return false;
        }

        // 첫 시도이거나 backoff 시간이 지났으면 재시도
        if (this.lastRetryAt == null) {
            return true;
        }

        long elapsed = java.time.Duration.between(this.lastRetryAt, LocalDateTime.now())
            .toMillis();
        return elapsed >= getBackoffMillis();
    }

    public void resetToPending() {
        if (this.status == OrderOutboxStatus.PUBLISHING) {
            this.status = OrderOutboxStatus.PENDING;
        }
    }

}

🔹 이 클래스의 역할 (왜 존재하는가)

이 코드는 Order 서비스에서 Outbox Pattern을 구현한 엔티티로, “주문 트랜잭션과 이벤트 발행을 분리하면서도 정합성을 보장”하기 위한 핵심 구성요소다. 전체 구조를 기준 → 필드 → 상태 전이 → 재시도/백오프 전략 순서로 알아봤다.

OrderOutboxEvent는 다음 목적을 가진다.

• Order DB 트랜잭션 안에서 이벤트를 안전하게 저장
• Kafka / 메시지 브로커 발행 실패 시에도 이벤트 유실 방지
• 비동기 재시도 + 지수 백오프를 통해 안정적인 이벤트 발행 보장
• Saga / 이벤트 기반 아키텍처에서 Exactly-once에 가까운 동작 기반

즉,

“주문은 성공했는데 이벤트는 못 보냈다” → 절대 발생하지 않게 만드는 구조

🔹 테이블 / 엔티티 구조

@Entity
@Table(name = "p_order_outbox_events")
public class OrderOutboxEvent extends BaseEntity

• p_order_outbox_events

• 일반 도메인 엔티티가 아니라 이벤트 로그 / 메시지 큐 대체 테이블

• BaseEntity:

  • 보통 createdAt, updatedAt 같은 audit 필드 포함

🔹 핵심 필드 알아보기

🔸 식별자 및 이벤트 기본 정보

@Id
@GeneratedValue(strategy = GenerationType.UUID)
private UUID id;

• Outbox 이벤트 자체의 ID
• orderId와 1:N 관계 가능 (한 주문에서 여러 이벤트)

@Column(nullable = false)
private UUID orderId;

• 어떤 주문에서 발생한 이벤트인지 식별

@Column(nullable = false)
private String eventType;

• 예: OrderAfterCreateV1, OrderCanceledV1
• Consumer가 분기 처리할 수 있는 기준값

@Column(columnDefinition = "TEXT", nullable = false)
private String payload;

• Kafka로 보낼 JSON 직렬화 데이터
• 스키마 변경에도 유연 (TEXT)

🔸 상태 관리 필드 (Outbox의 핵심)

@Enumerated(EnumType.STRING)
private OrderOutboxStatus status;

일반적인 상태 흐름:

PENDING → PUBLISHING → PUBLISHED
                   ↘ FAILED

private Integer retryCount = 0;

• 발행 시도 횟수
• 백오프 계산의 기준값

private LocalDateTime publishedAt;
private LocalDateTime lastRetryAt;

• 성공 시각 / 마지막 시도 시각 기록
• 모니터링, 운영 분석에 매우 중요

🔹 객체 생성 방식 (팩토리 메서드)

public static OrderOutboxEvent create(UUID orderId, String eventType, String payload)

의도:

• 무조건 PENDING 상태로 시작
• 생성 규칙을 한 곳에서 통제
• 생성자 노출 차단 (protected)

👉 트랜잭션 내부에서 다음처럼 사용됨

OrderOutboxEvent event =
    OrderOutboxEvent.create(orderId, "OrderAfterCreateV1", payload);
outboxRepository.save(event);

🔹 재시도 / 백오프 정책 (매우 중요한 부분)

🔸 재시도 제한 상수

private static final int MAX_RETRY_COUNT = 20;
private static final long INITIAL_BACKOFF_MILLIS = 1000L;
private static final long MAX_BACKOFF_MILLIS = 300000L;

• 최대 20회 시도
• 1초 → 2초 → 4초 → … → 최대 5분
• AWS / Google Cloud 권장 전략과 동일

🔸상태 전이 메서드

발행 시도 시작

public void markAsPublishing() {
    this.status = PUBLISHING;
    this.lastRetryAt = now;
    this.retryCount++;
}

• 실제 Kafka 전송 직전에 호출
• 중복 실행 방지용 락 역할

발행 성공

public void markAsPublished()

• 최종 성공 상태
• 이후 절대 재시도하지 않음

발행 실패

public void markAsFailed()

• retryCount 증가
• 20회 초과 시 FAILED 확정
• 운영자가 수동 개입해야 하는 상태

🔹 재시도 가능 여부 판단

public boolean canRetry()

• PUBLISHED / FAILED → ❌
• PENDING / PUBLISHING → ⭕

public boolean shouldRetryNow()

• 최초 시도 → 즉시 가능

• 이후:

  • 현재 시간 - lastRetryAt >= backoffMillis

👉 스케줄러가 이 메서드만 믿고 동작 가능

🔹 지수 백오프 + Jitter 구현

long exponentialBackoff =
    1초 * 2^(retryCount - 1);

double jitterFactor = 0.8 ~ 1.2;

왜 Jitter를 넣었나?

• 여러 이벤트가 동시에 재시도되는 Thundering Herd 방지
• 클라우드 환경에서 권장되는 방식

이 구현은 실무 기준에서도 매우 잘 짜인 편이다. (권장되는 방식)

🔹 resetToPending()의 의미

public void resetToPending()

사용 시점:

• PUBLISHING 상태에서

  • 앱 크래시
  • 트랜잭션 롤백
  • 락 타임아웃

같은 상황이 발생했을 때

👉 다시 잡아서 재처리 가능하도록 복구

🔹 이 엔티티가 보장하는 것

이 설계로 인해 보장되는 특성:

🔹 한 줄 요약

이 OrderOutboxEvent는 주문 서비스에서 Kafka 이벤트 발행을 “운영 수준으로 안전하게” 만드는 Outbox 엔티티이며, 지수 백오프 + 재시도 + 상태 머신이 결합된 실전형 구현이다.

Outbox / Inbox 패턴을 “깊이(depth)” 단계별로 쌓아 올리는 방식으로 정리해봤다. (이벤트 기반/마이크로서비스에서 “DB 트랜잭션 ↔ 메시지 발행”의 정합성을 맞추는 용도)

🔹 depth.1 이론(왜 필요한가 / 무엇을 해결하나)

🔸 Outbox 패턴

• 서비스가 DB에 상태 변경(예: 주문 생성)을 한 뒤, 곧바로 Kafka 같은 브로커로 이벤트를 발행하면 DB 커밋은 성공했는데 메시지 발행은 실패(또는 반대) 같은 “반쪽 성공”이 생길 수 있다.

• Outbox는 이를 막기 위해,

  1. 업무 데이터 변경과
  2. 이벤트 기록(Outbox 테이블 insert) 를 같은 DB 트랜잭션으로 묶고, 이후 별도 프로세스(퍼블리셔)가 Outbox를 읽어 브로커로 발행한다.

즉 “DB가 진실(Source of Truth)이고, 이벤트는 DB에 먼저 적어둔 뒤 안전하게 발행”하는 구조.

🔸 Inbox 패턴

• 브로커(Kafka)는 “최소 1회(at-least-once)” 전달이 흔해서, 컨슈머는 중복 메시지를 받을 수 있다.

• Inbox는 컨슈머가 메시지를 처리할 때,

  • 메시지의 고유 식별자(eventId/messageId)를 기준으로
  • 이미 처리했는지(Inbox 테이블/스토어로) 확인하고
  • 중복이면 무시(멱등) 하도록 보장한다.

즉 “수신 측에서 중복 처리를 방지해서 정확히 한 번처럼 보이게” 만드는 구조.

🔹 depth.2 최소 구성요소

🔸 Outbox 최소 구성

1. Outbox 테이블

   • id(eventId, UUID 등), aggregate_type, aggregate_id, event_type
   • payload(JSON), status(PENDING/SENT/FAILED), created_at

2. 업무 트랜잭션 안에서 Outbox 레코드 저장

   • 예: 주문 저장 + outbox insert를 같은 트랜잭션으로

3. Outbox 퍼블리셔(폴링 워커)

   • 주기적으로 status=PENDING 레코드를 조회 → Kafka 발행 → 성공 시 SENT로 업데이트

4. 재시도 정책(최소)

   • 실패하면 FAILED로 남기고 다음 폴링에서 재시도(또는 backoff)

최소 구성에서는 “폴링 기반”이 가장 단순

🔸 Inbox 최소 구성

1. Inbox 테이블(Processed Messages)

   • event_id(유니크), consumer_name, processed_at

2. 컨슘 시 처리 순서

   • 트랜잭션 시작
   • event_id가 Inbox에 이미 있으면 → 종료(중복 무시)
   • 없으면 → 업무 로직 처리(상태 변경) + Inbox insert
   • 커밋

핵심은 “업무 처리”와 “처리완료 기록”을 같은 트랜잭션으로 묶는 것.

🔹 depth.3 도입 시 추가로 신경 써야 하는 것

🔸 이벤트 “키”와 “중복” 설계

• Inbox가 있으려면 eventId가 반드시 고유해야 하고, 컨슈머가 그걸 신뢰해야 한다.

• eventId 생성 주체

  • Outbox 레코드 id를 eventId로 쓰는 게 가장 깔끔(발행되는 메시지에 그대로 포함)

🔸 Outbox 퍼블리셔의 동시성/락

• 퍼블리셔가 2개 이상 떠도 안전해야 함(이중 발행 방지)

• 방법(선택지)

  • SELECT … FOR UPDATE SKIP LOCKED
  • 상태를 PENDING -> SENDING으로 CAS 업데이트 후 발행
  • 배치 단위로 “claim” 후 처리

🔸 발행 성공 판정과 브로커 ack

• “Kafka 발행 성공”을 어디까지로 볼지

  • 프로듀서 ack(브로커에 기록 완료)까지 받으면 성공 처리

• 실패/타임아웃 시: 실제론 기록됐는데 타임아웃일 수 있어 → 중복 발행 가능성을 전제로 Inbox가 필요해짐(서로 보완 관계)

🔸 메시지 순서 보장(aggregate 단위)

• 주문 같은 aggregate는 “생성→상태변경” 순서가 중요함
• Kafka를 쓴다면 partition key를 aggregateId(orderId) 로 고정해서 순서가 깨지지 않게

🔸 스키마 진화와 호환성

• payload(JSON)의 필드가 바뀌면 컨슈머가 깨지기 쉬움

• 최소 방어:

  • event_type + version 넣기(OrderCreatedV1 같은 네이밍)
  • 컨슈머는 모르는 필드는 무시할 수 있게(유연한 역직렬화)

🔸 장애/모니터링/운영

• Outbox에 FAILED가 쌓이면 “업무는 됐는데 이벤트가 안 나감” 상태

• 최소한 필요:

  • FAILED 건수 알람
  • 오래된 PENDING 감지
  • DLQ(Dead Letter) 또는 수동 재처리 도구

🔹 depth.4 확장 시 추천 체크리스트

• CDC 기반 Outbox (Debezium) 폴링 대신 DB binlog를 읽어 Kafka로 내보내 성능/지연 개선

• Exactly-once처럼 보이게 만들기 조합

  • 생산: Outbox(또는 Kafka 트랜잭션)
  • 소비: Inbox + 멱등 업데이트(유니크 제약/업서트)

• Idempotency Key를 도메인 명령에도 적용

  • “주문 생성 요청” 자체도 중복 호출될 수 있으면 API 레벨에도 idempotency key 추가

• 리플레이 전략

  • Inbox 보관 기간(TTL) 결정: 영구 보관 vs 일정 기간 후 정리

• 대용량 최적화

  • outbox/inbox 파티셔닝, 아카이빙, payload 분리(큰 payload는 object storage + pointer)

🔹 현재 내 프로젝트 기준으로 봤을 때

• order-server

  • 주문 생성 트랜잭션에 Outbox 기록: OrderAfterCreateV1

• hub-server

  • OrderAfterCreateV1 컨슘 시 Inbox로 중복 제거
  • 허브경로 계산 결과 저장 + Outbox로 HubRouteAfterCreateV1

• delivery-server

  • HubRouteAfterCreateV1 컨슘 시 Inbox
  • HubDelivery/FirmDelivery 생성(트랜잭션)

보상 트랜잭션(Compensating Transaction)은 분산 환경에서 이미 커밋된 작업을 “되돌리기 위해” 실행하는 추가 트랜잭션을 말한다.

❌ “롤백”이 아니라 ✅ “취소·무효·상쇄”를 새로운 트랜잭션으로 수행하는 개념이다.

🔹 왜 보상 트랜잭션이 필요한가?

🔸 단일 DB 트랜잭션

BEGIN
  주문 생성
  결제 처리
ROLLBACK  ← 문제 생기면 한 번에 되돌림

• ACID 보장
• 하나의 DB / 하나의 트랜잭션 매니저
• 롤백 가능

🔸 분산 트랜잭션 (MSA)

order-server      → 주문 생성 (COMMIT)
payment-server    → 결제 승인 (COMMIT)
delivery-server   → 배송 생성 ❌ 실패

• 이미 각 서비스는 COMMIT 완료
• 전통적인 ROLLBACK 불가능
• 2PC는 현실적으로 성능·가용성 문제 큼

👉 그래서 등장한 개념이 보상 트랜잭션

🔹 보상 트랜잭션의 정의

보상 트랜잭션이란 이미 성공적으로 커밋된 트랜잭션의 효과를 비즈니스적으로 무효화하기 위해 실행하는 “역방향 트랜잭션”이다.

🔹 실제 예시로 보면 가장 명확함

예: 주문 → 배송 생성 실패

🔸 정상 흐름

1. 주문 생성 ✅
2. 허브 경로 계산 ✅
3. 배송 생성 ❌ 실패

🔸 이때 “보상”은?

• 주문을 ROLLBACK ❌ (이미 커밋됨)
• 대신:

주문 상태 → CANCELED
배송 예약 데이터 → 취소 처리
결제 → 환불 (필요 시)

이 전체가 보상 트랜잭션이다.

🔹 Saga 패턴에서의 위치

보상 트랜잭션은 Saga 패턴의 핵심 구성 요소다.

🔸 Choreography Saga

• 실패 이벤트 발생
• 각 서비스가 자기 보상 로직 수행

🔸 Orchestrator Saga

• 중앙 Orchestrator가
• “A 실패 → B 보상 → C 보상” 순서 제어

👉 어떤 Saga든 보상 트랜잭션 없이는 성립 불가

🔹 보상 트랜잭션의 중요한 특성

🔸 항상 “완전 복구”를 목표로 하지 않는다

• 재고 복구 ❌
• 주문 취소 상태로 마킹 ✅
• 금전적 환불만 보장하는 경우도 많음

🔸 멱등성(Idempotency)이 필수

같은 보상 이벤트가 2번 와도
결과는 1번 수행된 것과 동일해야 함

• Kafka 중복 소비
• 네트워크 재시도
• 장애 복구 시 재처리

🔸 순서 보장이 중요

• 배송 취소 → 결제 환불 (순서 중요)
• 잘못하면 “배송 완료 후 환불” 같은 사고 발생

🔸 보상도 실패할 수 있다

• 그래서 재시도 / DLQ / 알림 필요
• “보상 트랜잭션도 하나의 비즈니스 트랜잭션”

🔹 요약 정리

보상 트랜잭션은 분산 환경에서 이미 커밋된 작업을 기술적으로 롤백할 수 없을 때, 비즈니스적으로 상태를 상쇄하기 위해 수행하는 역방향 트랜잭션이며, Saga 패턴의 핵심 구성 요소다.

🔹 order-server

🔸 OrderStatusChangedV1 DTO 추가

public record OrderStatusChangedV1(
    UUID orderId,
    OrderStatus orderStatus,
    java.time.LocalDateTime changedAt
) {}

🔸 OrderCommandService

현재 구현된 코드를 기준으로 봤을 때, 이미

@Transactional
    public void deleteOrder(UUID id) {
}

• 여기서 취소 + 재고복원 + soft delete까지 책임지고 있으니, 여기에 추가로 OrderStatusChangedV1를 같이 발행하도록 하면 흐름이 깔끔하다.

• OrderStatusChangedV1 이벤트를 발행하는 위치는 order.updateStatus(CANCELED) + order.delete() + 재고복원까지 다 성공한 다음, 마지막에 발행하는 것이 좋다. 이유는 주문 취소가 확정일 때만 보상 처리하도록 하는 게 일관성에 좋기 때문이다.

• 여기서 멱등성: deleteOrder()가 두 번 호출되면? 상황도 생각해봤을 때, Order.updateStatus()는 COMPLETED, CANCELED -> false라서 이미 CANCELED면 INVALID_ORDER_STATUS_TRANSITION이 터질 것이다.

• 이후에 트랜잭션 커밋 전에 이벤트가 나가면 생길 문제를 방지하기 위해, (Kafka 전송은 성공했는데 DB 커밋이 실패하면 delivery-server는 “취소됐다고 믿고 배송 취소”를 해버릴 수 있기 때문) @TransactionalEventListener(phase = AFTER_COMMIT)로 커밋 이후에 Kafka 발행하도록 처리방식으로 가벼운 고도화 또 가능. ➡️ 지금은 일단 보류

@Transactional
    public void deleteOrder(UUID id) {

        // 주문 조회
        Order order = readOrderOrThrow(id);

        // 담당 허브 소속 주문인지 체크
        if (SecurityUtils.hasRole(Role.HUB_MANAGER)) {
            List<UUID> managingHubId = hubPort.readHubId(SecurityUtils.getCurrentUserId());
            UUID receiverHubId = firmPort.readHubId(order.getReceiverFirmId());

            if (!managingHubId.contains(receiverHubId)) {
                throw new BusinessException(ErrorCode.ORDER_NOT_IN_MANAGING_HUB);
            }
        }

        // 본인 주문인지 체크
        if (SecurityUtils.hasRole(Role.FIRM_MANAGER)) {
            UUID currentUserId = SecurityUtils.getCurrentUserId();

            if (!currentUserId.equals(order.getCreatedBy())) {
                throw new BusinessException(ErrorCode.ORDER_NOT_CREATED_BY_USER);
            }
        }

        order.updateStatus(OrderStatus.CANCELED);
        order.delete(SecurityUtils.getCurrentUserId());

        // 재고 복원
        for (OrderProduct p : order.getOrderProductList()) {
            productPort.recoverStock(p.getProductId(), p.getQuantity());
        }

        // Kafka 메시지 발행
        OrderStatusChangedV1 message = new OrderStatusChangedV1(
            order.getId(),
            order.getOrderStatus(),  // CANCELED
            LocalDateTime.now()
        );

        eventPublisher.sendOrderStatusChanged(message);

        // TODO: 주문 읽기 업데이트 (OrderStatus, delete)
    }

🔸 EventPublisher

package chill_logistics.order_server.domain.event;

import chill_logistics.order_server.application.dto.command.OrderAfterCreateV1;

public interface EventPublisher {

    void sendOrderAfterCreate(OrderAfterCreateV1 message);

    /* 여기 추가 */
    void sendOrderStatusChanged(OrderStatusChangedV1 message);
}

🔸 OrderAfterCreateProducer

OrderAfterCreateProducer 클래스명 OrderEventPublisher로 수정하면 더 깔끔할 것 같다.

package chill_logistics.order_server.infrastructure.kafka;

import chill_logistics.order_server.domain.event.EventPublisher;
import chill_logistics.order_server.application.dto.command.OrderAfterCreateV1;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderAfterCreateProducer implements EventPublisher {

    private final KafkaTemplate<String, OrderAfterCreateV1> orderAfterCreateKafkaTemplate;
    private final KafkaTemplate<String, OrderStatusChangedV1> orderStatusChangedKafkaTemplate;

    @Value("${app.kafka.topic.order-after-create}")
    private String orderAfterCreateTopic;

    @Value("${app.kafka.topic.order-status-changed}")
    private String orderStatusChangedTopic;

    /**
     * 주문 생성 후 Kafka로 OrderAfterCreate 이벤트를 발행합니다.
     *
     * @param message 주문 생성 정보를 담고 있는 메시지 객체
     */
    @Override
    public void sendOrderAfterCreate(OrderAfterCreateV1 message) {

        String key = message.orderId().toString();

        log.info("[Kafka] OrderAfterCreate 메시지 발행, topic={}, key={}, message={}",
                orderAfterCreateTopic, key, message);

        orderAfterCreateKafkaTemplate
                .send(orderAfterCreateTopic, key, message)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("[Kafka] OrderAfterCreate 메시지 전송 실패, key={}", key, ex);
                    } else {
                        log.info("[Kafka] OrderAfterCreate 메시지 전송 성공, orderId={}, offset={}",
                                key, result.getRecordMetadata().offset());
                    }
                });
    }

    /* 여기 추가 */

    /**
     * 주문 상태 변경(취소/실패/완료 등) 시 Kafka로 OrderStatusChanged 이벤트 발행
     */
    @Override
    public void sendOrderStatusChanged(OrderStatusChangedV1 message) {

        String key = message.orderId().toString();

        log.info("[Kafka] OrderStatusChanged 메시지 발행, topic={}, key={}, message={}",
            orderStatusChangedTopic, key, message);

        orderStatusChangedKafkaTemplate
            .send(orderStatusChangedTopic, key, message)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("[Kafka] OrderStatusChanged 메시지 전송 실패, key={}", key, ex);
                } else {
                    log.info("[Kafka] OrderStatusChanged 메시지 전송 성공, orderId={}, offset={}",
                        key, result.getRecordMetadata().offset());
                }
            });
    }
}

🔸 application.yml

app:
  kafka:
    topic:
      order-after-create: order-after-create
      /* 여기 추가 */
      order-status-changed: order-status-changed

🔸 KafkaProducerConfig

package chill_logistics.order_server.infrastructure.config;

import chill_logistics.order_server.application.dto.command.OrderAfterCreateV1;
import chill_logistics.order_server.application.dto.command.OrderStatusChangedV1;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

    /* 공통 Producer 설정 */
    private Map<String, Object> baseProducerProps() {
        Map<String, Object> props = new HashMap<>();

        // Kafka Broker
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 직렬화 설정
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return props;
    }

    /* OrderAfterCreate */
    @Bean
    public ProducerFactory<String, OrderAfterCreateV1> orderAfterCreateProducerFactory() {
        return new DefaultKafkaProducerFactory<>(baseProducerProps());
    }

    @Bean
    public KafkaTemplate<String, OrderAfterCreateV1> orderAfterCreateKafkaTemplate() {
        return new KafkaTemplate<>(orderAfterCreateProducerFactory());
    }

    /* OrderStatusChanged */
    @Bean
    public ProducerFactory<String, OrderStatusChangedV1> orderStatusChangedProducerFactory() {
        return new DefaultKafkaProducerFactory<>(baseProducerProps());
    }

    @Bean
    public KafkaTemplate<String, OrderStatusChangedV1> orderStatusChangedKafkaTemplate() {
        return new KafkaTemplate<>(orderStatusChangedProducerFactory());
    }
}

🔹 delivery-server

• HubRouteAfterCreate 처리(배송 생성) 쪽에서 “이미 취소된 주문이면 생성하지 않기” 같은 가드 로직 필요

• 멱등성: orderId 기준으로 이미 배송 만들어졌으면, 다시 생성하지 않도록 처리 필요 (같은 주문으로 배송이 중복 생성되지 않도록)

🔸 HubDelievery

package chill_logistics.delivery_server.domain.entity;

import ...

    ...(생략)

    // length=15 옵션 제거
    @Enumerated(EnumType.STRING)
    @Column(name = "delivery_status", nullable = false)
    private DeliveryStatus deliveryStatus;

    ...(생략)

    ...(기존 메서드 생략)

    // 주문 취소 시 처리 메서드
    public void cancelDueToOrder() {

        // 멱등성: 이미 취소된 상태면 다시 호출되어도 OK || 이미 상품 받았으면 주문 취소 불가
        if (this.deliveryStatus == DeliveryStatus.DELIVERY_CANCELLED
            || this.deliveryStatus == DeliveryStatus.DELIVERY_COMPLETED) {
            return;
        }

        DeliveryStatus nextDeliveryStatus = DeliveryStatus.DELIVERY_CANCELLED;

        if (!this.deliveryStatus.canTransitTo(nextDeliveryStatus)) {
            throw new BusinessException(ErrorCode.DELIVERY_ALREADY_COMPLETED_OR_CANCELED);
        }

        this.deliveryStatus = nextDeliveryStatus;
    }
}

🔸 FirmDelivery

package chill_logistics.delivery_server.domain.entity;

import ...

...(생략)

    // length=15 옵션 제거
    @Enumerated(EnumType.STRING)
    @Column(name = "delivery_status", nullable = false)
    private DeliveryStatus deliveryStatus;

    ...(생략)

    ... (기존 메서드 생략)

    // 주문 취소 시 처리 메서드
    public void cancelDueToOrder() {

        // 멱등성: 이미 취소된 상태면 다시 호출되어도 OK || 이미 상품 받았으면 주문 취소 불가
        if (this.deliveryStatus == DeliveryStatus.DELIVERY_CANCELLED
            || this.deliveryStatus == DeliveryStatus.DELIVERY_COMPLETED) {
            return;
        }

        DeliveryStatus nextDeliveryStatus = DeliveryStatus.DELIVERY_CANCELLED;

        if (!this.deliveryStatus.canTransitTo(nextDeliveryStatus)) {
            throw new BusinessException(ErrorCode.DELIVERY_ALREADY_COMPLETED_OR_CANCELED);
        }

        this.deliveryStatus = nextDeliveryStatus;
    }
}

🔸 OrderStatus 추가

package chill_logistics.delivery_server.application;

public enum OrderStatus {

    CREATED,
    PROCESSING,
    IN_TRANSIT,
    COMPLETED,
    CANCELED
}

🔸 OrderStatusChangedV1 추가

package chill_logistics.delivery_server.infrastructure.kafka.dto;

import chill_logistics.delivery_server.application.OrderStatus;
import java.time.LocalDateTime;
import java.util.UUID;

public record OrderStatusChangedV1(
    UUID orderId,
    OrderStatus orderStatus,
    String reason,
    LocalDateTime changedAt
) {}

🔸 KafkaConsumerConfig

• 공통 설정 분리 후 HubRouteAfterCreate / OrderStatusChanged 각각 container 설정

package chill_logistics.delivery_server.infrastructure.config;

import chill_logistics.delivery_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import chill_logistics.delivery_server.infrastructure.kafka.dto.OrderStatusChangedV1;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

@Configuration
public class KafkaConsumerConfig {

    /* 공통 Consumer 설정 */
    private Map<String, Object> baseConsumerProps(String groupId) {

        Map<String, Object> props = new HashMap<>();

        // Kafka Broker
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Consumer Group ID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // Key 역직렬화
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // Value 역직렬화
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        // 처음 실행 시 offset 정책 (earliest: consumer group이 이 토픽의 이 파티션을 지금 들어오는 새 메시지부터 읽는다)
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        return props;
    }

    /* HubRouteAfterCreate */
    @Bean
    public ConsumerFactory<String, HubRouteAfterCreateV1> hubConsumerFactory() {

        // JSON → HubRouteAfterCreateV1 역직렬화를 위한 Deserializer
        JsonDeserializer<HubRouteAfterCreateV1> deserializer =
            new JsonDeserializer<>(HubRouteAfterCreateV1.class, false);

        // Kafka 메시지 역직렬화 시 허용할 패키지를 명시적으로 지정
        deserializer.addTrustedPackages(
            "chill_logistics.delivery_server.infrastructure.kafka.dto"
        );

        // 위 설정값 + Key/Value Deserializer를 기반으로 실제 Consumer 인스턴스를 만들어 KafkaListener에 제공
        return new DefaultKafkaConsumerFactory<>(
            baseConsumerProps("delivery-server-hub-route-group"),
            new StringDeserializer(),   // Key Deserializer (String)
            deserializer                // Value Deserializer (HubRouteAfterCreateV1)
        );
    }

    /* HubRouteAfterCreate */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, HubRouteAfterCreateV1>
    // @KafkaListener가 사용할 Listener 컨테이너 생성 (ConsumerFactory를 주입하여 실제 Kafka Consumer 동작을 구성)
    hubKafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, HubRouteAfterCreateV1> factory =
            new ConcurrentKafkaListenerContainerFactory<>();

        // ConsumerFactory 설정
        factory.setConsumerFactory(hubConsumerFactory());

        return factory;
    }

    /* OrderStatusChanged */
    @Bean
    public ConsumerFactory<String, OrderStatusChangedV1> orderStatusChangedConsumerFactory() {

        JsonDeserializer<OrderStatusChangedV1> deserializer =
            new JsonDeserializer<>(OrderStatusChangedV1.class, false);

        deserializer.addTrustedPackages(
            "chill_logistics.delivery_server.infrastructure.kafka.dto",
            "chill_logistics.delivery_server.application"  // OrderStatus ENUM 역직렬화용
        );

        return new DefaultKafkaConsumerFactory<>(
            baseConsumerProps("delivery-server-order-status-group"),
            new StringDeserializer(),
            deserializer
        );
    }

    /* OrderStatusChanged */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderStatusChangedV1>
    orderStatusChangedKafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, OrderStatusChangedV1> factory =
            new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(orderStatusChangedConsumerFactory());

        return factory;
    }
}

🔸 OrderStatusChangedListener 추가

package chill_logistics.delivery_server.infrastructure.kafka;

import chill_logistics.delivery_server.application.OrderStatus;
import chill_logistics.delivery_server.application.service.OrderCancellationService;
import chill_logistics.delivery_server.infrastructure.kafka.dto.OrderStatusChangedV1;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderStatusChangedListener {

    private final OrderCancellationService orderCancellationService;

    @KafkaListener(
        topics = "order-status-changed",
        groupId = "delivery-server-order-status-group",
        containerFactory = "orderStatusChangedKafkaListenerContainerFactory"
    )
    public void listen(OrderStatusChangedV1 message) {

        log.info("Kafka 메시지 수신: {}", message);

        if (message.orderStatus() == OrderStatus.CANCELED) {
            orderCancellationService.cancelDeliveriesByOrder(
                message.orderId()
            );
        }
    }
}

🔸 OrderCancellationService 추가

package chill_logistics.delivery_server.application.service;

import chill_logistics.delivery_server.domain.entity.FirmDelivery;
import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.FirmDeliveryRepository;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import java.util.List;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderCancellationService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;

    /* [주문 취소/실패 시 배송 취소 메서드]
     * orderId 기준 HubDelivery / FirmDelivery 취소 처리
     */
    @Transactional
    public void cancelDeliveriesByOrder(UUID orderId) {

        List<HubDelivery> hubDeliveryList = hubDeliveryRepository.findByOrderId(orderId);
        List<FirmDelivery> firmDeliveryList = firmDeliveryRepository.findByOrderId(orderId);

        if (hubDeliveryList.isEmpty() && firmDeliveryList.isEmpty()) {
            log.info("[주문 취소에 따른 배송 취소] 대상 배송이 없습니다. orderId={}", orderId);

            return;
        }

        for (HubDelivery hubDelivery : hubDeliveryList) {
            hubDelivery.cancelDueToOrder();
        }

        for (FirmDelivery firmDelivery : firmDeliveryList) {
            firmDelivery.cancelDueToOrder();
        }

        log.info("[주문 취소에 따른 배송 취소] 처리 완료. orderId={}", orderId);
    }
}

🔸 1. 주문 취소 시 트랜잭션 추가

delivery-server가 정상 Flow의 가장 마지막에 위치하기 때문에, 배송 서비스의 보상 트랜잭션 작업을 하게 되면 다른 서버에서도 추가 작업은 필수로 요하게 된다.

때문에, 타 서비스의 기능 고도화 작업이 이루어지는 동안, delivery-server에서는 주문 취소 시의 Flow에 대한 트랜잭션을 추가하는 작업부터 하기로 결정.

작업 방향:

order-server에서 주문 취소 이벤트 추가 발행 → delivery-server에서 해당 이벤트 구독

🔸 2. 주문 생성 후 배송 생성 실패 시 보상 트랜잭션

order-server 성공 → hub-server 성공 → delivery-server 실패 이 시나리오의 보상 트랜잭션 작업 예정.

이 때 꼭 가져가야 할 것이 outbox/inbox이다. outbox에서 kafka 이벤트를 발행하고, 실패 시에도 주기적인 retry를 시도할 수 있다.

다만, 이렇게 되면 모든 서비스에서 추가적인 작업이 필요해지기 때문에, 팀원들과 의논 후 개발 일정을 정하는 것이 좋겠다고 판단된다.

🔹 요약 정리

➡️ 결론: order-server쪽 동시성 제어 하는 동안, case1의 보상 트랜잭션부터 마련하고 있자. 다음 과정으로 추후에 같이 outbox/inbox 도입하는 것이 베스트

case 3의 경우 추후에 작업 및 블로깅 예정

📌 기존의 성공 예상 시나리오:

• order-server: 주문 생성 (DB insert) → Kafka로 OrderAfterCreate 이벤트 발행
  
• hub-server: OrderAfterCreate 구독 → 허브 간 경로 알고리즘 + 예상 소요시간 계산 → Kafka로 HubRouteAfterCreate 이벤트 발행
  
• delivery-server: HubRouteAfterCreate 구독 → HubRouteAfterCreate에 들어있는 order/hub/firm 정보 + 예상 소요시간 기반 HubDelivery/FirmDelivery 생성

📌 실패 시 보상 예상 시나리오: 주문은 생성되어 위의 플로우를 다 돌았는데, 나중에 결제 실패 혹은 유저가 주문 취소해서 orderStatus가 FAILED 또는 CANCELED로 바뀔 때

• order-server: orderStatus FAILED/CANCELED로 변경 → OrderStatusChanged 이벤트 발행
  
• delivery-server: OrderStatusChanged 구독 → 이미 생성된 HubDelivery/FirmDelivery 취소 처리

< 각 이벤트 역할 > OrderAfterCreate → 배송 “생성용” 메시지 OrderStatusChangedV1 → 배송 “보상/취소용” 메시지

보상 트랜잭션용 추가 이벤트가 발행되어야 한다.

“주문 + 배송 생성”을 하나의 비즈니스 트랜잭션으로 봤을 때 order-server 쪽 실패를 delivery-server에서 어떻게 되돌릴지를 설계하는 거니까, 사실상 작은 Saga 설계라고 생각하면서 고민해봤다.

아래 순서로 정리해보면:

1. 어디서 실패가 나냐에 따라 케이스 나누기
2. 간단하면서도 실용적인 보상 전략 (이벤트 기반 보상)
3. 필수로 넣어야 할 것들: 상태 모델, 이벤트, outbox, idempotency
4. 조금 더 진하게 가고 싶으면: Orchestrator 기반 설계 방향

🔹 실패 위치별로 케이스 먼저 정리

현재 그림(대략):

1. order-server에서 주문 생성
2. OrderAfterCreateV1 Kafka 발행
3. hub-server에서 허브 경로 계산 후 HubRouteAfterCreateV1 발행
4. delivery-server에서 HubDelivery / FirmDelivery 생성

여기서 “order-server가 실패했다”를 조금 더 쪼개면:

🔸 (A) 주문 생성 트랜잭션 안에서 실패 (DB insert도 안 됨)

• 예: 유효성 검증 실패, DB 제약조건 위반 등으로 주문 자체가 롤백.
• 이 경우엔 애초에 OrderAfterCreate 이벤트가 발행되지 않도록 트랜잭션을 묶어야 함. ⇒ 보상 트랜잭션 필요 X (이벤트 안 나갔으니 delivery-server도 아무것도 안 함)

🔸 (B) 주문은 생성됐는데, 이후 단계에서 비즈니스적으로 실패

대표 예:

• 결제 실패
• 사장님이 주문 직후 “바로 취소”
• 재고 체크 후 불가 판정

이 경우에는 이미:

• OrderAfterCreate가 발행되어,
• 허브 경로 계산 + HubRouteAfterCreate
• delivery-server에서 HubDelivery/FirmDelivery까지 만들어졌을 가능성이 높음.

👉 이 상황에서 보상 트랜잭션 = “이미 만들어진 배송을 취소/무효화” 하는 것.

그래서 내가 설계해야 할 건 주로 (B) 케이스다.

🔹 가장 현실적인 전략: “주문 상태 이벤트 기반 보상”

핵심 아이디어는 간단해:

“주문이 실패/취소로 바뀌면, 그걸 이벤트로 흘려보내고, delivery-server는 그걸 받아서 관련 배송들을 전부 ‘취소’로 바꾼다.”

🔸 주문 상태 모델 정리

order-server에서:

public enum OrderStatus {
    CREATED,     // 생성 완료 (아직 결제 전 혹은 준비 상태)
    CONFIRMED,   // 결제/검증까지 OK
    FAILED,      // 비즈니스적으로 실패 (결제 실패 등)
    CANCELED     // 사용자 취소
}

그리고 주문 상태가 FAILED, CANCELED로 전이될 때마다 이벤트를 발생:

• OrderFailedV1
• OrderCanceledV1

이벤트 payload 예:

public record OrderStatusChangedV1(
    UUID orderId,
    OrderStatus status,
    String reason
) {}

혹은 이벤트 타입을 분리해도 되고, status 필드로 한 번에 처리해도 됨.

🔸 delivery-server에서의 보상 트랜잭션

delivery-server는 이미 orderId 기준으로 HubDelivery/FirmDelivery를 가지고 있으니,

그럼 간단하게:

@Service
@RequiredArgsConstructor
public class OrderCompensationService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;

    @Transactional
    public void compensateDeliveriesForOrder(UUID orderId, String reason) {
        // 1) 허브 배송 취소
        List<HubDelivery> hubDeliveries = hubDeliveryRepository.findByOrderId(orderId);
        hubDeliveries.forEach(hd -> hd.cancel(reason)); // 엔티티 메서드에서 상태 변경 + 로그 남기기

        // 2) 업체 배송 취소
        List<FirmDelivery> firmDeliveries = firmDeliveryRepository.findByOrderId(orderId);
        firmDeliveries.forEach(fd -> fd.cancel(reason));
    }
}

그리고 Kafka Consumer 예시:

@KafkaListener(topics = "order-status-changed")
public void handleOrderStatusChanged(OrderStatusChangedV1 event) {
    if (event.status() == OrderStatus.FAILED || event.status() == OrderStatus.CANCELED) {
        compensationService.compensateDeliveriesForOrder(event.orderId(), event.reason());
    }
}

🔸 배송 상태 Enum에 “취소” 추가

DeliveryStatus에 최소 이 정도:

public enum DeliveryStatus {
    CREATED,        // 생성됨
    READY,          // 출고 준비
    IN_TRANSIT,     // 이동 중
    COMPLETED,      // 배송 완료
    CANCELLED       // 주문 취소/실패로 인한 취소
}

그리고 엔티티 메서드:

public void cancel(String reason) {
    if (this.status == DeliveryStatus.IN_TRANSIT || this.status == DeliveryStatus.COMPLETED) {
        // 이미 너무 진행된 건 어떻게 할지 정책 결정 필요
        return;
    }
    this.status = DeliveryStatus.CANCELLED;
    this.cancellationReason = reason; // 컬럼 추가
}

정책 포인트

• 이미 IN_TRANSIT 이상이면 어떻게 할지?

  • CANCEL_REQUESTED 같은 중간 상태를 둘지,
  • 아니면 그냥 무시할지(“이미 나간 건 도로 못 가져온다”).

• 이건 도메인 정책이니까, 나중에 고도화 단계에서 결정해도 됨.

🔹 이 때 꼭 챙겨야 할 것들

🔸 Outbox 패턴 (order-server 쪽)

“주문 row는 생성됐는데, 이벤트 발행이 실패하면?” → 또 다른 일관성 깨짐.

그래서 order-server에서는:

1. 주문 테이블 insert/update
2. 같은 트랜잭션에서 outbox 테이블에 메시지 row insert
3. 별도 Outbox Publisher가 outbox 테이블에서 읽어서 Kafka로 발행 후, 성공 시 outbox row 삭제

이렇게 하면:

• DB에 있는 상태(주문 상태)와
• 밖으로 나가는 이벤트 를 최소한 같은 DB 트랜잭션으로 보장할 수 있음.

delivery-server 쪽도 중요 이벤트라면 동일하게 outbox 써도 좋고.

🔸 보상 로직의 “멱등성(idempotency)”

Kafka는 기본적으로 “at-least-once” 쪽이라 OrderStatusChangedV1가 중복으로 올 수 있음.

그러면 cancel() 여러 번 호출해도 문제가 없어야 함:

• status가 이미 CANCELLED이면 그냥 return
• 이벤트 처리할 때 이벤트 ID 기준으로 처리 여부 기록하는 것도 한 방법

간단하게는:

public void cancel(String reason) {
    if (this.status == DeliveryStatus.CANCELLED) return;
    ...
}

이 정도만 해도 MVP/고도화 1단계에서는 충분함.

🔹 “조금 더 제대로” 하고 싶을 때: Orchestrator 기반

지금 구조는 사실상 Choreography Saga에 가깝다:

• order-server → 이벤트 → hub-server → 이벤트 → delivery-server

근데 나중에:

• “주문 생성 + 허브 경로 계산 + 배송 생성 + AI + Slack/Discord” 이렇게 플로우가 더 복잡해지면,
• orchestrator-service를 하나 두고 중앙에서 단계별로 명령/보상하는 방식도 고려할 수 있다.

예를 들면:

sequenceDiagram
    participant Orchestrator
    participant Order as order-server
    participant Delivery as delivery-server

    Orchestrator->>Order: createOrder()
    Order-->>Orchestrator: OrderCreated(orderId)

    Orchestrator->>Delivery: createDeliveries(orderId)
    Delivery-->>Orchestrator: DeliveriesCreated

    Orchestrator->>Orchestrator: Saga 완료 처리

    %% 실패 케이스
    Orchestrator->>Delivery: createDeliveries(orderId)
    Delivery-->>Orchestrator: CreateFailed
    Orchestrator->>Order: compensateOrder(orderId)  // 주문 상태 FAILED/CANCELED로

이 패턴에서는:

• 각 스텝마다 “정방향 작업” + “보상 작업”을 쌍으로 정의

  • 예: createDeliveries ↔ cancelDeliveries
  • 예: createOrder ↔ markOrderFailed

• Orchestrator가 상태 머신처럼 “어디까지 성공했는지” 기억하고, 중간에 실패하면 역순으로 보상 호출.

다만, 지금은 Kafka 이벤트 기반 설계로 많이 가고 있어서 “주문 상태 이벤트 기반 보상(위 2번 방안)”을 먼저 적용하고, 나중에 진짜 복잡해졌을 때 orchestrator-service로 승격하는 구조가 현실적인 단계 같다.

🔹 정리

1. 진짜로 보상 필요한 구간은 “주문은 살아 있는데, 배송이 이미 생성된 후 주문이 실패/취소되는 구간”이다.

2. 그때는 order-server를 진실의 근원(Single Source of Truth)로 두고, 주문이 FAILED/CANCELED로 바뀔 때마다 이벤트 → delivery-server에서 관련 배송 전체 CANCELLED로 만드는 보상 로직을 두는 게 가장 깔끔.

3. 이를 위해:

   • 주문/배송 상태 Enum 정리
   • OrderStatusChanged 이벤트 정의
   • delivery-server 보상 서비스 구현
   • outbox + idempotency 챙기기

4. 나중에 플로우가 더 복잡해지면 orchestrator-service로 승격해서 진짜 Saga 상태 머신을 도입하면 됨.

package chill_logistics.delivery_server.application;

import chill_logistics.delivery_server.application.dto.command.AssignedDeliveryPersonV1;
import chill_logistics.delivery_server.application.dto.command.HubRouteAfterCommandV1;
import chill_logistics.delivery_server.application.dto.command.HubRouteHubInfoV1;
import chill_logistics.delivery_server.domain.entity.DeliveryStatus;
import chill_logistics.delivery_server.domain.entity.FirmDelivery;
import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.FirmDeliveryRepository;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import chill_logistics.delivery_server.presentation.ErrorCode;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryCancelRequestV1;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryStatusChangeRequestV1;
import java.util.List;
import java.util.UUID;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class DeliveryCommandService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;
    private final AsyncAiService asyncAiService;
    private final DeliveryPersonAssignmentService deliveryPersonAssignmentService;

    /* [허브 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 허브 배송 1 row 생성
     * pathHubIds를 기반으로 구간 계산 후 여러 번 호출 → N row 생성
     */
    @Transactional
    public void createHubDelivery(
        HubRouteAfterCommandV1 message,
        UUID segmentStartHubId,
        String segmentStartHubName,
        String segmentStartHubFullAddress,
        UUID segmentEndHubId,
        String segmentEndHubName,
        String segmentEndHubFullAddress,
        Integer expectedDeliveryDuration,  // 첫 구간만 값, 나머지는 null
        UUID hubDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[허브 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.WAITING_FOR_HUB;

        // HubDelivery 엔티티 생성
        HubDelivery hubDelivery = HubDelivery.createFromSegment(
            message,
            segmentStartHubId,
            segmentStartHubName,
            segmentStartHubFullAddress,
            segmentEndHubId,
            segmentEndHubName,
            segmentEndHubFullAddress,
            expectedDeliveryDuration,
            hubDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        HubDelivery savedHubDelivery = hubDeliveryRepository.save(hubDelivery);

        log.info("[허브 배송 생성 완료] hubDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedHubDelivery.getId(),
            savedHubDelivery.getOrderId(),
            savedHubDelivery.getDeliverySequenceNum());
    }

    /* [업체 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 업체 배송 생성
     */
    @Transactional
    public void createFirmDelivery(
        HubRouteAfterCommandV1 message,
        UUID firmDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[업체 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.MOVING_TO_FIRM;

        List<HubRouteHubInfoV1> pathHubs = message.pathHubs();

        if (pathHubs == null || pathHubs.isEmpty()) {
            throw new BusinessException(ErrorCode.INVALID_HUB_ROUTE_PATH);
        }

        HubRouteHubInfoV1 lastHub = pathHubs.get(pathHubs.size() - 1);
        UUID endHubId = lastHub.hubId();

        // FirmDelivery 엔티티 생성
        FirmDelivery firmDelivery = FirmDelivery.createFrom(
            message,
            endHubId,
            firmDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        // 업체 배송 저장
        FirmDelivery savedFirmDelivery = firmDeliveryRepository.save(firmDelivery);

        log.info("[업체 배송 생성 완료] firmDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedFirmDelivery.getId(),
            savedFirmDelivery.getOrderId(),
            savedFirmDelivery.getDeliverySequenceNum());
    }

    /* [전체 배송 생성]
     * pathHubs 기반 허브 구간 (row) 여러 개 생성 + 업체 배송 생성 = 전체 배송 생성
     * 전체 배송 생성 + AI + Discord 비동기 호출
     */
    @Transactional
    public void createDelivery(HubRouteAfterCommandV1 message) {

        log.info("[배송 생성 시작] orderId={}", message.orderId());

        List<HubRouteHubInfoV1> pathHubs = message.pathHubs();

        if (pathHubs == null || pathHubs.size() < 2) {
            throw new BusinessException(ErrorCode.INVALID_HUB_ROUTE_PATH);
        }

        // pthHubs 기반 허브 구간 수 계산
        int hubSegmentCount = pathHubs.size() - 1;

        log.info("[허브 구간 수 계산] orderId={}, hubSegmentCount={}", message.orderId(), hubSegmentCount);

        // 업체 배송 담당자
        UUID lastHubId = pathHubs.get(pathHubs.size() - 1).hubId();
        AssignedDeliveryPersonV1 firmDeliveryPerson =
            deliveryPersonAssignmentService.assignFirmDeliveryPerson(lastHubId);

        UUID firmDeliveryPersonId = firmDeliveryPerson.userId();

        // AI 메시지에 넣어줄 첫 구간 허브배송 담당자 이름
        String firstHubDeliveryPersonName = null;

        // 허브 구간 수 만큼 HubDelivery row 생성
        for (int i = 0; i < hubSegmentCount; i++) {

            // 1부터 시작
            int hubDeliverySequenceNum = i + 1;

            // 첫 허브 구간에만 총 예상 소요시간 기록, 나머지는 null
            Integer segmentExpectedDeliveryDuration =
                (i == 0) ? message.expectedDeliveryDuration() : null;

            HubRouteHubInfoV1 startHub = pathHubs.get(i);
            HubRouteHubInfoV1 endHub = pathHubs.get(i + 1);

            // 각 구간 시작 허브 기준으로 허브배송 담당자 배정
            AssignedDeliveryPersonV1 hubDeliveryPerson =
                deliveryPersonAssignmentService.assignHubDeliveryPerson(startHub.hubId());

            UUID hubDeliveryPersonId = hubDeliveryPerson.userId();
            String hubDeliveryPersonName = hubDeliveryPerson.userName();

            // 첫 구간 담당자 이름은 따로 저장 (AI용)
            if (i == 0) {
                firstHubDeliveryPersonName = hubDeliveryPersonName;
            }

            createHubDelivery(
                message,
                startHub.hubId(),
                startHub.hubName(),
                startHub.hubFullAddress(),
                endHub.hubId(),
                endHub.hubName(),
                endHub.hubFullAddress(),
                segmentExpectedDeliveryDuration,
                hubDeliveryPersonId,
                hubDeliverySequenceNum
            );
        }

        // 업체 배송 sequenceNum = 마지막 허브 구간 + 1
        int firmDeliverySequenceNum = hubSegmentCount + 1;

        createFirmDelivery(message, firmDeliveryPersonId, firmDeliverySequenceNum);

        log.info(
            "[배송 생성 완료 & 배송 담당자 배정 완료] orderId={}", message.orderId());

        // AI + Discord 비동기 체인 호출
        asyncAiService.sendDeadlineRequest(message, firstHubDeliveryPersonName);
    }

    /* [배송 상태 변경]
     * 허브 배송 / 업체 배송 상태 변경
     */
    @Transactional
    public void changeDeliveryStatus(UUID deliveryId, DeliveryStatusChangeRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[허브 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[업체 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }
    }

    /* [배송 취소]
     * 허브 배송 / 업체 배송 취소
     */
    @Transactional
    public void cancelDelivery(UUID deliveryId, DeliveryCancelRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.cancelDelivery();

            log.info("[허브 배송 취소] deliveryId={}, orderId={}", deliveryId, hubDelivery.getOrderId());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.cancelDelivery();

            log.info("[업체 배송 취소] deliveryId={}, orderId={}", deliveryId, firmDelivery.getOrderId());
        }
    }

    private HubDelivery getHubDeliveryByIdOrThrow(UUID hubDeliveryId) {

        return hubDeliveryRepository.findById(hubDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.HUB_DELIVERY_NOT_FOUND));
    }

    private FirmDelivery getFirmDeliveryByIdOrThrow(UUID firmDeliveryId) {

        return firmDeliveryRepository.findById(firmDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.FIRM_DELIVERY_NOT_FOUND));
    }
}

🔸 허브 배송 단일 row 생성 방식

package chill_logistics.delivery_server.application;

import chill_logistics.delivery_server.application.dto.command.AssignedDeliveryPersonV1;
import chill_logistics.delivery_server.application.dto.command.HubRouteAfterCommandV1;
import chill_logistics.delivery_server.domain.entity.DeliveryStatus;
import chill_logistics.delivery_server.domain.entity.FirmDelivery;
import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.FirmDeliveryRepository;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import chill_logistics.delivery_server.presentation.ErrorCode;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryCancelRequestV1;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryStatusChangeRequestV1;
import java.util.List;
import java.util.UUID;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class DeliveryCommandService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;
    private final AsyncAiService asyncAiService;
    private final DeliveryPersonAssignmentService deliveryPersonAssignmentService;

    /* [허브 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 허브 배송 생성
     */
    @Transactional
    public void createHubDelivery(
        HubRouteAfterCommandV1 message,
        UUID hubDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[허브 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.WAITING_FOR_HUB;

        // HubDelivery 엔티티 생성
        HubDelivery hubDelivery = HubDelivery.createFrom(
            message,
            hubDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        // 허브 배송 저장
        HubDelivery savedHubDelivery = hubDeliveryRepository.save(hubDelivery);

        log.info("[허브 배송 생성 완료] hubDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedHubDelivery.getId(),
            savedHubDelivery.getOrderId(),
            savedHubDelivery.getDeliverySequenceNum());
    }

    /* [업체 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 업체 배송 생성
     */
    @Transactional
    public void createFirmDelivery(
        HubRouteAfterCommandV1 message,
        UUID firmDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[업체 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.MOVING_TO_FIRM;

        // FirmDelivery 엔티티 생성
        FirmDelivery firmDelivery = FirmDelivery.createFrom(
            message,
            firmDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        // 업체 배송 저장
        FirmDelivery savedFirmDelivery = firmDeliveryRepository.save(firmDelivery);

        log.info("[업체 배송 생성 완료] firmDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedFirmDelivery.getId(),
            savedFirmDelivery.getOrderId(),
            savedFirmDelivery.getDeliverySequenceNum());
    }

    /* [전체 배송 생성]
     * 허브 배송 + 업체 배송 = 전체 배송 생성
     * 전체 배송 생성 + AI + Discord 비동기 호출
     */
    @Transactional
    public void createDelivery(HubRouteAfterCommandV1 message) {

        log.info("[배송 생성 시작] orderId={}", message.orderId());

        // 허브 배송 담당자 배정
        AssignedDeliveryPersonV1 hubDeliveryPerson =
            deliveryPersonAssignmentService.assignHubDeliveryPerson();

        // 업체 배송 담당자 배정
        AssignedDeliveryPersonV1 firmDeliveryPerson =
            deliveryPersonAssignmentService.assignFirmDeliveryPerson();

        UUID hubDeliveryPersonId = hubDeliveryPerson.userId();
        String hubDeliveryPersonName = hubDeliveryPerson.userName();
        UUID firmDeliveryPersonId = firmDeliveryPerson.userId();

        // pathHubIds 기반 deliverySequenceNum 계산
        int hubDeliverySequenceNum = calculateHubSequenceNum(message.pathHubIds());
        int firmDeliverySequenceNum = hubDeliverySequenceNum + 1;

        createHubDelivery(message, hubDeliveryPersonId, hubDeliverySequenceNum);
        createFirmDelivery(message, firmDeliveryPersonId, firmDeliverySequenceNum);

        log.info(
            "[배송 생성 완료 & 배송 담당자 배정 완료] orderId={}, hubDeliveryPersonId={}, firmDeliveryPersonId={}",
            message.orderId(), hubDeliveryPersonId, firmDeliveryPersonId);

        // AI + Discord 비동기 체인 호출
        asyncAiService.sendDeadlineRequest(message, hubDeliveryPersonName);
    }

    /* [배송 상태 변경]
     * 허브 배송 / 업체 배송 상태 변경
     */
    @Transactional
    public void changeDeliveryStatus(UUID deliveryId, DeliveryStatusChangeRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[허브 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[업체 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }
    }

    /* [배송 취소]
     * 허브 배송 / 업체 배송 취소
     */
    @Transactional
    public void cancelDelivery(UUID deliveryId, DeliveryCancelRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.cancelDelivery();

            log.info("[허브 배송 취소] deliveryId={}, orderId={}", deliveryId, hubDelivery.getOrderId());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.cancelDelivery();

            log.info("[업체 배송 취소] deliveryId={}, orderId={}", deliveryId, firmDelivery.getOrderId());
        }
    }

    private HubDelivery getHubDeliveryByIdOrThrow(UUID hubDeliveryId) {

        return hubDeliveryRepository.findById(hubDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.HUB_DELIVERY_NOT_FOUND));
    }

    private FirmDelivery getFirmDeliveryByIdOrThrow(UUID firmDeliveryId) {

        return firmDeliveryRepository.findById(firmDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.FIRM_DELIVERY_NOT_FOUND));
    }

    /* [pathHubIds 기반 허브 배송 sequenceNum 계산 메서드]
     * 허브 경로가 순서대로 들어있는 pathHybIds [hub1, hub2, hub3] 가 있는 경우
     * 허브 구간 수 = 2 (hub1→hub2, hub2→hub3)
     * 허브 간 이동 구간 수 = pathHubIds.size() - 1
     * 허브 간 이동 구간 수 없거나 1개 이하면 1로 처리
     * 업체 배송 시퀀스 번호 = hubSequence + 1 (항상 허브 배송 뒤의 마지막 단계이므로)
     */
    private int calculateHubSequenceNum(List<UUID> pathHubIds) {

        // 허브 정보 없거나, 경유 허브 없는 경우 최소 1
        if (pathHubIds == null || pathHubIds.size() < 2) {
            return 1;
        }

        // 허브 간 이동 구간 수 = 노드 수 - 1
        return pathHubIds.size() - 1;
    }
}

🔸 허브 배송 1 row → N row 생성 방식

package chill_logistics.delivery_server.application;

import chill_logistics.delivery_server.application.dto.command.AssignedDeliveryPersonV1;
import chill_logistics.delivery_server.application.dto.command.HubRouteAfterCommandV1;
import chill_logistics.delivery_server.application.dto.command.HubRouteHubInfoV1;
import chill_logistics.delivery_server.domain.entity.DeliveryStatus;
import chill_logistics.delivery_server.domain.entity.FirmDelivery;
import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.FirmDeliveryRepository;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import chill_logistics.delivery_server.presentation.ErrorCode;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryCancelRequestV1;
import chill_logistics.delivery_server.presentation.dto.request.DeliveryStatusChangeRequestV1;
import java.util.List;
import java.util.UUID;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class DeliveryCommandService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;
    private final AsyncAiService asyncAiService;
    private final DeliveryPersonAssignmentService deliveryPersonAssignmentService;

    /* [허브 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 허브 배송 1 row 생성
     * pathHubIds를 기반으로 구간 계산 후 여러 번 호출 → N row 생성
     */
    @Transactional
    public void createHubDelivery(
        HubRouteAfterCommandV1 message,
        UUID segmentStartHubId,
        String segmentStartHubName,
        String segmentStartHubFullAddress,
        UUID segmentEndHubId,
        String segmentEndHubName,
        String segmentEndHubFullAddress,
        Integer expectedDeliveryDuration,  // 첫 구간만 값, 나머지는 null
        UUID hubDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[허브 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.WAITING_FOR_HUB;

        // HubDelivery 엔티티 생성
        HubDelivery hubDelivery = HubDelivery.createFromSegment(
            message,
            segmentStartHubId,
            segmentStartHubName,
            segmentStartHubFullAddress,
            segmentEndHubId,
            segmentEndHubName,
            segmentEndHubFullAddress,
            expectedDeliveryDuration,
            hubDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        HubDelivery savedHubDelivery = hubDeliveryRepository.save(hubDelivery);

        log.info("[허브 배송 생성 완료] hubDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedHubDelivery.getId(),
            savedHubDelivery.getOrderId(),
            savedHubDelivery.getDeliverySequenceNum());
    }

    /* [업체 배송 생성 메서드]
     * Kafka 메시지로 order 정보 받아와서 업체 배송 생성
     */
    @Transactional
    public void createFirmDelivery(
        HubRouteAfterCommandV1 message,
        UUID firmDeliveryPersonId,
        int deliverySequenceNum) {

        log.info("[업체 배송 생성 시작] orderId={}, deliverySequenceNum={}",
            message.orderId(), deliverySequenceNum);

        // 초기 배송 상태 셋팅
        DeliveryStatus deliveryStatus = DeliveryStatus.MOVING_TO_FIRM;

        List<HubRouteHubInfoV1> pathHubs = message.pathHubs();

        if (pathHubs == null || pathHubs.isEmpty()) {
            throw new BusinessException(ErrorCode.INVALID_HUB_ROUTE_PATH);
        }

        HubRouteHubInfoV1 lastHub = pathHubs.get(pathHubs.size() - 1);
        UUID endHubId = lastHub.hubId();

        // FirmDelivery 엔티티 생성
        FirmDelivery firmDelivery = FirmDelivery.createFrom(
            message,
            endHubId,
            firmDeliveryPersonId,
            deliverySequenceNum,
            deliveryStatus
        );

        // 업체 배송 저장
        FirmDelivery savedFirmDelivery = firmDeliveryRepository.save(firmDelivery);

        log.info("[업체 배송 생성 완료] firmDeliveryId={}, orderId={}, deliverySequenceNum={}",
            savedFirmDelivery.getId(),
            savedFirmDelivery.getOrderId(),
            savedFirmDelivery.getDeliverySequenceNum());
    }

    /* [전체 배송 생성]
     * pathHubs 기반 허브 구간 (row) 여러 개 생성 + 업체 배송 생성 = 전체 배송 생성
     * 전체 배송 생성 + AI + Discord 비동기 호출
     */
    @Transactional
    public void createDelivery(HubRouteAfterCommandV1 message) {

        log.info("[배송 생성 시작] orderId={}", message.orderId());

        // 허브 배송 담당자 배정
        AssignedDeliveryPersonV1 hubDeliveryPerson =
            deliveryPersonAssignmentService.assignHubDeliveryPerson();

        // 업체 배송 담당자 배정
        AssignedDeliveryPersonV1 firmDeliveryPerson =
            deliveryPersonAssignmentService.assignFirmDeliveryPerson();

        UUID hubDeliveryPersonId = hubDeliveryPerson.userId();
        String hubDeliveryPersonName = hubDeliveryPerson.userName();
        UUID firmDeliveryPersonId = firmDeliveryPerson.userId();

        List<HubRouteHubInfoV1> pathHubs = message.pathHubs();

        if (pathHubs == null || pathHubs.size() < 2) {
            throw new BusinessException(ErrorCode.INVALID_HUB_ROUTE_PATH);
        }

        // pthHubs 기반 허브 구간 수 계산
        int hubSegmentCount = pathHubs.size() - 1;

        log.info("[허브 구간 수 계산] orderId={}, hubSegmentCount={}", message.orderId(), hubSegmentCount);

        // 허브 구간 수 만큼 HubDelivery row 생성
        for (int i = 0; i < hubSegmentCount; i++) {

            // 1부터 시작
            int hubDeliverySequenceNum = i + 1;

            // 첫 허브 구간에만 총 예상 소요시간 기록, 나머지는 null
            Integer segmentExpectedDeliveryDuration =
                (i == 0) ? message.expectedDeliveryDuration() : null;

            HubRouteHubInfoV1 startHub = pathHubs.get(i);
            HubRouteHubInfoV1 endHub = pathHubs.get(i + 1);

            createHubDelivery(
                message,
                startHub.hubId(),
                startHub.hubName(),
                startHub.hubFullAddress(),
                endHub.hubId(),
                endHub.hubName(),
                endHub.hubFullAddress(),
                segmentExpectedDeliveryDuration,
                hubDeliveryPersonId,
                hubDeliverySequenceNum
            );
        }

        // 업체 배송 sequenceNum = 마지막 허브 구간 + 1
        int firmDeliverySequenceNum = hubSegmentCount + 1;

        createFirmDelivery(message, firmDeliveryPersonId, firmDeliverySequenceNum);

        log.info(
            "[배송 생성 완료 & 배송 담당자 배정 완료] orderId={}, hubDeliveryPersonId={}, firmDeliveryPersonId={}",
            message.orderId(), hubDeliveryPersonId, firmDeliveryPersonId);

        // AI + Discord 비동기 체인 호출
        asyncAiService.sendDeadlineRequest(message, hubDeliveryPersonName);
    }

    /* [배송 상태 변경]
     * 허브 배송 / 업체 배송 상태 변경
     */
    @Transactional
    public void changeDeliveryStatus(UUID deliveryId, DeliveryStatusChangeRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[허브 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.changeStatus(request.nextDeliveryStatus());
            log.info("[업체 배송 상태 변경] deliveryId={}, nextDeliveryStatus={}",
                deliveryId, request.nextDeliveryStatus());
        }
    }

    /* [배송 취소]
     * 허브 배송 / 업체 배송 취소
     */
    @Transactional
    public void cancelDelivery(UUID deliveryId, DeliveryCancelRequestV1 request) {

        if (request.deliveryType() == DeliveryType.HUB) {
            HubDelivery hubDelivery = getHubDeliveryByIdOrThrow(deliveryId);

            hubDelivery.cancelDelivery();

            log.info("[허브 배송 취소] deliveryId={}, orderId={}", deliveryId, hubDelivery.getOrderId());
        }

        if (request.deliveryType() == DeliveryType.FIRM) {
            FirmDelivery firmDelivery = getFirmDeliveryByIdOrThrow(deliveryId);

            firmDelivery.cancelDelivery();

            log.info("[업체 배송 취소] deliveryId={}, orderId={}", deliveryId, firmDelivery.getOrderId());
        }
    }

    private HubDelivery getHubDeliveryByIdOrThrow(UUID hubDeliveryId) {

        return hubDeliveryRepository.findById(hubDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.HUB_DELIVERY_NOT_FOUND));
    }

    private FirmDelivery getFirmDeliveryByIdOrThrow(UUID firmDeliveryId) {

        return firmDeliveryRepository.findById(firmDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.FIRM_DELIVERY_NOT_FOUND));
    }
}

비교적 간단한 roundRobin 방식으로 먼저 구현을 해보았다.

package chill_logistics.delivery_server.application;

import chill_logistics.delivery_server.application.dto.command.AssignedDeliveryPersonV1;
import chill_logistics.delivery_server.infrastructure.user.UserClient;
import chill_logistics.delivery_server.infrastructure.user.dto.UserForDeliveryResponseV1;
import chill_logistics.delivery_server.presentation.ErrorCode;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class DeliveryPersonAssignmentService {

    private static final String ROLE_HUB_DELIVERY_PERSON = "HUB_DELIVERY_PERSON";
    private static final String ROLE_FIRM_DELIVERY_PERSON = "FIRM_DELIVERY_PERSON";

    private final UserClient userClient;

    // 역할별 라운드 로빈 인덱스 저장 (멀티스레드 환경 대응)
    private final Map<String, AtomicLong> roundRobinIndex = new ConcurrentHashMap<>();

    // 허브 배송 담당자 배정
    public AssignedDeliveryPersonV1 assignHubDeliveryPerson() {
        return assignByRole(ROLE_HUB_DELIVERY_PERSON);
    }

    // 업체 배송 담당자 배정
    public AssignedDeliveryPersonV1 assignFirmDeliveryPerson() {
        return assignByRole(ROLE_FIRM_DELIVERY_PERSON);
    }

    private AssignedDeliveryPersonV1 assignByRole(String role) {

        // user-server에서 role 별 유저 목록 조회
        List<UserForDeliveryResponseV1> deliveryCandidates = userClient.getUsersByRole(role);

        if (deliveryCandidates == null || deliveryCandidates.isEmpty()) {
            throw new BusinessException(ErrorCode.DELIVERT_PERSON_NOT_FOUND);
        }

        // 해당 role에 대한 라운드 로빈 인덱스 가져오고, 없으면 생성
        AtomicLong counter = roundRobinIndex.computeIfAbsent(role, key -> new AtomicLong(0L));

        // 현재 인덱스를 가져오고 증가시키는 연산
        long index = counter.incrementAndGet();

        // 후보자 수로 나머지 연산하려 실제 선택 위치 계산
        int selectedIndex = (int) (index % deliveryCandidates.size());

        // 최종 선택된 담당자
        UserForDeliveryResponseV1 chosen = deliveryCandidates.get(selectedIndex);

        return new AssignedDeliveryPersonV1(
            chosen.userId(),
            chosen.userName()
        );
    }
}

📌 시나리오:

• DeliveryCommandService 가 허브/업체 배송을 생성하고
• 동시에 AsyncAiService 를 통해
• OpenAiClient(AiClient 구현체) 에게 “이 주문, 언제까지 보내야 하는지 + Discord 메시지” 생성을 시키고
• 그 결과를 AiDeadlineResponseV1 로 받아서 (지금은 로그만 찍고, 나중에 Discord 서비스로 보낼 예정)
• 이 전체 흐름에서 프롬프트 템플릿은 OpenAiConstants.DISCORD_FORMATTED_DEADLINE_PROMPT 가 담당.

🔸 HubRouteAfterCommandV1 메시지 수신

Kafka 리스너로 HubRouteAfterCommandV1 라는 DTO를 들고 DeliveryCommandService.createDelivery(...) 를 호출.

🔸 DeliveryCommandService

DeliveryCommandService.createDelivery(...) 호출

   @Transactional
   public void createDelivery(
       HubRouteAfterCommandV1 message,
       UUID hubDeliveryPersonId,
       UUID firmDeliveryPersonId) {

       log.info("[배송 생성 시작] orderId={}", message.orderId());

       createHubDelivery(message, hubDeliveryPersonId);  // 허브 배송 생성
       createFirmDelivery(message, firmDeliveryPersonId); // 업체 배송 생성

       log.info("[배송 생성 완료] orderId={}", message.orderId());

       asyncAiService.sendDeadlineRequest(message); // ★ AI 비동기 호출
   }

• DB 트랜잭션 안에서 허브 배송 + 업체 배송 엔티티를 각각 생성 & 저장.
• 그 후 AsyncAiService.sendDeadlineRequest(message) 로 AI 호출을 비동기로 날림.

🔸 AsyncAiService

AsyncAiService.sendDeadlineRequest(...)

   @Async
   public void sendDeadlineRequest(HubRouteAfterCommandV1 message) {
       log.info("[AI 비동기 호출 시작] orderId={}", message.orderId());

       AiDeadlineRequestV1 request = new AiDeadlineRequestV1(
           message.orderId(),
           message.startHubId(),
           message.startHubName(),
           message.startHubFullAddress(),
           message.endHubId(),
           message.endHubName(),
           message.endHubFullAddress(),
           message.receiverFirmId(),
           message.receiverFirmFullAddress(),
           message.receiverFirmOwnerName(),
           message.requestNote(),
           message.productName(),
           message.productQuantity(),
           message.orderCreatedAt(),
           message.expectedDeliveryDuration()
       );

       AiDeadlineResponseV1 response = aiClient.generateDeadlineMessage(request);

       log.info("[AI 결과] orderId={}, finalDeadline={}", message.orderId(),response.finalDeadline());

       // TODO: Discord 서비스로 response.discordMessage() 전달
   }

• @Async 덕분에 별도의 스레드에서 실행 → 배송 생성 트랜잭션과 분리.
• HubRouteAfterCommandV1 → AiDeadlineRequestV1 로 변환 (AI에게 줄 데이터 정제).
• AiClient.generateDeadlineMessage(request) 호출 → 실제 OpenAI 호출은 구현체(OpenAiClient)가 담당.
• AI 응답으로 받은 AiDeadlineResponseV1 에서 finalDeadline 을 로그에 남김.
• 나중에는 response.discordMessage() 를 Discord 서비스에 보내서 실제 DM/알림으로 사용할 예정.

🔸 OpenAiClient

OpenAiClient.generateDeadlineMessage(...)

   @Override
   public AiDeadlineResponseV1 generateDeadlineMessage(AiDeadlineRequestV1 request) {

       String prompt = buildPrompt(request);

       log.info("[AI 요청] 배송 최종 발송 시한 계산, orderId={}", request.orderId());

       String outputText = chatClient
           .prompt()
           .user(prompt)
           .call()
           .content();

       log.info("[AI 응답 수신] 배송 최종 발송 시한 계산 완료, orderId={}", request.orderId());

       String finalDeadline = extractDeadline(outputText);

       return new AiDeadlineResponseV1(outputText, finalDeadline);
   }

• buildPrompt(request) 로 거대한 텍스트 프롬프트 생성.

• chatClient.prompt().user(prompt).call().content() 로 Spring AI의 ChatClient 를 통해 OpenAI 호출.

• 전체 응답(outputText)에서 extractDeadline(...) 으로 "yyyy-MM-dd HH:mm" 형식의 최종 발송 시한만 파싱.

• AiDeadlineResponseV1 에

  • discordMessage: Discord 코드블록 그대로
  • finalDeadline: 파싱된 마지막 줄 시간만 담아서 리턴.

🔸 OpenAiConstants

OpenAiConstants.DISCORD_FORMATTED_DEADLINE_PROMPT

• AI에게 전달하는 전체 템플릿 문자열.
• 예시 1, 2, 3 을 포함해서 “이런 형식으로 Discord 코드블록을 만들어라” 를 강하게 가이드.
• 마지막에 실제 주문 정보 위치에 "%s" / "%d" 자리로 템플릿 변수를 만들어 두고,
• OpenAiClient.buildPrompt(...) 에서 formatted(...) 호출로 실제 값 채워 넣음.

🔹 각 클래스의 역할

🔸 OpenAiConstants

• 순수 상수 모음 클래스.

• 생성자를 private 으로 막아서 인스턴스 생성 불가.

• 현재 역할:

  • Discord용 메시지 + 발송 시한 계산 규칙 전체를 담은 프롬프트 템플릿 제공.
  • 예시 1, 2, 3 포함.
  • 마지막 부분에 실제 주문 정보 %s, %d 자리 (formatted()로 채움)를 제공.

🔸 AiClient (인터페이스)

public interface AiClient {
    AiDeadlineResponseV1 generateDeadlineMessage(AiDeadlineRequestV1 request);
}

• AI 호출을 추상화한 인터페이스.

• 장점:

  • 나중에 OpenAI가 아니라 다른 모델/다른 구현체 로 교체해도

    • AsyncAiService 입장에서는 AiClient 만 보면 됨.

  • 테스트 시에는 FakeAiClient / StubAiClient 를 주입하기 쉬움.

🔸 OpenAiClient (AiClient 구현체)

• AiClient 의 실제 구현체, Spring Bean (@Component).

• ChatClient (Spring AI)를 주입받아서 OpenAI와 통신.

• buildPrompt(...):

  • request 를 보고

    • requestNote 가 비어 있으면 "없음" 으로 치환.
    • 경유지(route)는 아직 "없음" 으로 고정 (TODO 주석 있음).

  • OpenAiConstants.DISCORD_FORMATTED_DEADLINE_PROMPT.formatted(...) 로 실제 주문 값 바인딩.

• extractDeadline(...):

  • 줄 단위로 쪼갠 뒤
  • "위 내용을 기반으로 도출된 최종 발송 시한은" 으로 시작하는 줄을 찾아서
  • 앞/뒤 고정 문구를 제거 → "yyyy-MM-dd HH:mm" 부분만 남김.
  • 찾지 못하면 null.

🔸 AsyncAiService

• 애플리케이션 레벨에서 AI 호출을 담당하는 비동기 서비스.

• 역할:

  1. HubRouteAfterCommandV1 → AiDeadlineRequestV1 로 매핑.
  2. AiClient 를 통해 실제 AI 호출.
  3. 결과 로그 남기고, 나중에 Discord 서비스로 메시지 전달 예정.

• @Async 덕분에:

  • DeliveryCommandService.createDelivery(...) 처리가 AI 응답을 기다리지 않고 바로 끝날 수 있음.
  • 배송 생성 로직과 AI 호출을 느슨하게 연결.

🔸 DeliveryCommandService

• 배송과 관련된 “명령(Command)” 책임을 한 데 모은 서비스.

• 주요 기능:

  • 배송 생성 + AI 호출 (createDelivery)
  • 배송 상태 변경 (changeDeliveryStatus)
  • 배송 취소 (cancelDelivery)

• createDelivery() 가 도메인 관점의 “전체 배송 생성” 유즈케이스:

  • HubDelivery + FirmDelivery 를 모두 생성.
  • 그다음에 AI에게 이 주문의 “최종 발송 시한 + Discord 메시지”를 비동기 요청.

고속 대용량 로그/이벤트를 토픽 단위로 쌓아두고, 여러 서비스가 비동기로 구독해서 처리하는 분산 메시징/스트리밍 플랫폼.

조금 자세히 보면:

🔸 핵심 개념

1. Broker

   • Kafka 서버 인스턴스 하나.
   • 여러 대를 띄워서 클러스터를 구성.

2. Topic

   • 메시지를 쌓아두는 “채널” 이름.
   • 예: order-created, payment-completed, user-signup 등.

3. Partition

   • Topic을 물리적으로 쪼개놓은 단위.
   • 병렬 처리(Consumer 여러 개)와 확장성(스케일 아웃)을 위해 존재.
   • 각 메시지는 파티션 안에서 순서를 가짐(Offset).

4. Producer

   • 메시지를 “보내는 쪽” 애플리케이션.
   • KafkaTemplate, REST API 서비스 등.

5. Consumer

   • 메시지를 “받아서 처리하는 쪽”.
   • 메시지 핸들링 로직을 가짐.

6. Consumer Group

   • 같은 그룹에 속한 Consumer들이 하나의 Topic 파티션들을 나눠서 병렬 처리.
   • 그룹 단위로 “한 메시지는 그룹 내에서 딱 한 Consumer만” 처리.

7. Offset

   • 파티션 내에서 메시지의 위치(일련 번호).
   • Consumer는 마지막으로 읽은 Offset을 기억해, 재시작 시 이어서 처리.

🔸 Kafka 왜 사용하는가?

• 느슨한 결합: 주문 서비스 → 결제 서비스 → 알림 서비스 간을 동기 HTTP 호출 대신, Kafka Topic으로 느슨하게 연결.
• 비동기 처리: 주문 요청에 바로 응답하고, 뒤에서 배송/알림/포인트 적립 처리.
• 버퍼 역할: 트래픽이 몰려도 Kafka가 일단 적재하고, Consumer가 천천히 따라잡을 수 있음.
• 재처리 가능: 특정 Offset부터 다시 읽어서 “replay” 가능 (이벤트 소싱/로그 분석에 좋음).

🔹 outBox

outbox는 kafka 메시지 전송에 실패했을 때에 대한 처리를 위해 필요하므로, 추후에 적용 예정.

outbox 역할:

• 실패했을 때 트래킹을 위해서 필요
• 지금은 Kafka로 바로 보내는데, 만약 처음부터 outbox 적용한다고 하면, order에서 createAfterOrder 보낼 때 outbox 테이블로 구성해서 여기에 메시지 정보를 저장해두고, 스케줄러가 outbox 데이터를 읽어서 주기적으로 보내주는 형태로 구현.
• outbox 적용함으로써 주기적인 retry 가능 (실패에 대한 fallback 역할)

지금은 MVP 개발에 집중하고, 고도화 기간에 실패했을 때 어떡하지?에 대한 처리로 적용하면 베스트

Spring Boot에서 Kafka 적용 예시

“주문 생성 이벤트”를 Kafka로 발행하고, 다른 서비스가 그걸 구독해서 처리하는 시나리오로 생각해보았다.

Docker Compose로 Kafka가 localhost:9092(기본 포트번호)로 떠 있다고 가정했을 때,

📌 시나리오

• order-server (Producer) : 주문 생성 API → Kafka OrderAfterCreate 이벤트 발행

• hub-server (Consumer) : OrderAfterCreate 구독 → 메시지에서 받아온 주문 정보 기반으로 알고리즘 적용 후 예상 소요시간 계산
• hub-server (Producer) : Kafka HubRouteAfterCrete(주문 정보 + 예상 소요 시간) 이벤트 발행

  • delivery-server (Consumer) : Kafka HubRouteAfterCrete 구독 → 배송 생성

🔹 order-server (Producer)

🔸 order-server: build.gradle

implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'

🔸 order-server: application.yml

spring:
  application:
    name: order-server

  kafka:
    bootstrap-servers: localhost:9092

app:
  kafka:
    topic:
      order-after-create: order-after-create

🔸 order-server: (infrastructure.config) KafkaProducerConfig

package chill_logistics.order_server.infrastructure.config;

import chill_logistics.order_server.infrastructure.kafka.dto.OrderAfterCreateV1;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, OrderAfterCreateV1> orderAfterCreateProducerFactory() {

        Map<String, Object> props = new HashMap<>();

        // Kafka Broker
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 직렬화 설정
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        // 필요시 기타 옵션 (acks, retries 등) 추가

        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<String, OrderAfterCreateV1> orderAfterCreateKafkaTemplate() {
        return new KafkaTemplate<>(orderAfterCreateProducerFactory());
    }
}

🔸 order-server: (infrastructure.kafka.dto) OrderAfterCreateV1

package chill_logistics.order_server.infrastructure.kafka.dto;

import java.time.LocalDateTime;
import java.util.UUID;

public record OrderAfterCreateV1(
    UUID orderId,
    UUID supplierHubId,
    UUID receiverHubId,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt
) {}

🔸 order-server: OrderAfterCreteProducer

package chill_logistics.order_server.infrastructure.kafka;

import chill_logistics.order_server.infrastructure.kafka.dto.OrderAfterCreateV1;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderAfterCreateProducer {

    private final KafkaTemplate<String, OrderAfterCreateV1> orderAfterCreateKafkaTemplate;

    @Value("${app.kafka.topic.order-after-create}")
    private String orderAfterCreateTopic;

    /* [주문 생성 후 Kafka로 OrderAfterCreate 이벤트 발행]
     */
    public void sendOrderAfterCreate(OrderAfterCreateV1 message) {

        String key = message.orderId().toString();

        log.info("[Kafka] OrderAfterCreate 메시지 발행, topic={}, key={}, message={}",
            orderAfterCreateTopic, key, message);

        orderAfterCreateKafkaTemplate
            .send(orderAfterCreateTopic, key, message)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("[Kafka] OrderAfterCreate 메시지 전송 실패, key={}", key, ex);
                } else {
                    log.info("[Kafka] OrderAfterCreate 메시지 전송 성공, orderId={}, offset={}",
                        key, result.getRecordMetadata().offset());
                }
            });
    }
}

🔸 order-server: OrderService에서 사용 예시

package chill_logistics.order_server.application;

import chill_logistics.order_server.domain.entity.Order;
import chill_logistics.order_server.domain.repository.OrderRepository;
import chill_logistics.order_server.infrastructure.kafka.OrderAfterCreateProducer;
import chill_logistics.order_server.infrastructure.kafka.dto.OrderAfterCreateV1;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OrderAfterCreateProducer orderAfterCreateProducer;

    @Transactional
    public UUID createOrder(CreateOrderRequest request) {

        // 1. 주문 엔티티 생성
        Order order = Order.create(
            request.supplierHubId(),
            request.receiverHubId(),
            request.receiverFirmId(),
            request.receiverFirmFullAddress(),
            request.receiverFirmOwnerName(),
            request.requestNote(),
            request.productName(),
            request.productQuantity()
        );

        // 2. 저장
        orderRepository.save(order);

        // 3. Kafka 메시지 생성
        OrderAfterCreateV1 message = new OrderAfterCreateV1(
            order.getId(),
            order.getSupplierHubId(),
            order.getReceiverHubId(),
            order.getReceiverFirmId(),
            order.getReceiverFirmFullAddress(),
            order.getReceiverFirmOwnerName(),
            order.getRequestNote(),
            order.getProductName(),
            order.getProductQuantity(),
            order.getCreatedAt()
        );

        // 4. Kafka 메시지 발행
        orderAfterCreateProducer.sendOrderAfterCreate(message);

        log.info("주문 생성 완료 + Kafka 메시지 발행 완료. orderId={}", order.getId());

        return order.getId();
    }
}

🔹 hub-server (Consumer)

🔸 hub-server: build.gradle

implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'

🔸 hub-server: application.yml

spring:
  application:
    name: hub-server

  kafka:
    bootstrap-servers: localhost:9092

🔸 hub-server: (infrastructure.config) KafkaConsumerConfig

package chill_logistics.hub_server.infrastructure.config;

import chill_logistics.hub_server.infrastructure.kafka.dto.OrderAfterCreateV1;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, OrderAfterCreateV1> orderConsumerFactory() {

        // JSON → OrderAfterCreateV1 역직렬화를 위한 Deserializer
        JsonDeserializer<OrderAfterCreateV1> deserializer =
            new JsonDeserializer<>(OrderAfterCreateV1.class, false);

        // Kafka 메시지 역직렬화 시 허용할 패키지를 명시적으로 지정
        deserializer.addTrustedPackages(
            "chill_logistics.delivery_server.infrastructure.kafka.dto"
        );

        // Kafka Consumer 설정 값
        Map<String, Object> properties = new HashMap<>();

        // Kafka Broker 주소 (Docker Compose에서 기본적으로 localhost:9092로 띄움)
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Consumer Group ID (같은 Group으로 묶인 Consumer들은 같은 메시지를 중복 처리하지 않음)
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "hub-server-group");

        // 메시지 Key 역직렬화 방식
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 메시지 Value 역직렬화 방식 (JsonDeserializer 사용)
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        // 위 설정값 + Key/Value Deserializer를 기반으로 실제 Consumer 인스턴스를 만들어 KafkaListener에 제공
        return new DefaultKafkaConsumerFactory<>(
            properties,
            new StringDeserializer(),   // Key Deserializer (String)
            deserializer                // Value Deserializer (OrderAfterCreateV1)
        );
    }

    // @KafkaListener가 사용할 Listener 컨테이너 생성 (ConsumerFactory를 주입하여 실제 Kafka Consumer 동작을 구성)
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderAfterCreateV1>
    orderKafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, OrderAfterCreateV1> factory =
            new ConcurrentKafkaListenerContainerFactory<>();

        // ConsumerFactory 설정
        factory.setConsumerFactory(orderConsumerFactory());

        return factory;
    }
}

🔸 hub-server: (infrastructure.kafka.dto) OrderAfterCreateV1

package chill_logistics.hub_server.infrastructure.kafka.dto;

import java.time.LocalDateTime;
import java.util.UUID;

public record OrderAfterCreateV1(
    UUID orderId,
    UUID supplierHubId,
    UUID receiverHubId,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt) {

    // application 계층의 command 변환 메서드
    public OrderAfterCommandV1 toCommand() {
        return new OrderAfterCommandV1(
            orderId(),
            supplierHubId(),
            receiverHubId(),
            receiverFirmId(),
            receiverFirmFullAddress(),
            receiverFirmOwnerName(),
            requestNote(),
            productName(),
            productQuantity(),
            orderCreatedAt(),
        );
    }
}

🔸 hub-server: (application.dto) OrderAfterCommandV1

package chill_logistics.hub_server.application.dto.command;

import java.time.LocalDateTime;
import java.util.UUID;

public record OrderAfterCommandV1(
    UUID orderId,
    UUID supplierHubId,
    UUID receiverHubId,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt,
) {}

🔸 hub-server: OrderAfterCreateListener

package chill_logistics.hub_server.infrastructure.kafka;

import chill_logistics.hub_server.application.HubCommandService;
import chill_logistics.hub_server.application.dto.command.OrderAfterCommandV1;
import chill_logistics.hub_server.infrastructure.kafka.dto.OrderAfterCreateV1;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderAfterCreateListener {

    private final HubCommandService hubCommandService;

    @KafkaListener(
        topics = "order-after-create",
        containerFactory = "orderKafkaListenerContainerFactory"
    )
    public void listen(OrderAfterCreateV1 message) {

        log.info("Kafka 메시지 수신: {}", message);

        // DTO → Command 변환
        OrderAfterCommandV1 command = message.toCommand();

        hubCommandService.calculateExpectedDuration(command);
    }
}

🔹 hub-server (Producer)

🔸 hub-server: build.gradle

implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'

🔸 hub-server: application.yml

spring:
  application:
    name: hub-server

  kafka:
    bootstrap-servers: localhost:9092

app:
  kafka:
    topic:
      hub-route-after-create: hub-route-after-create

🔸 hub-server: (infrastructure.config) KafkaProducerConfig

package chill_logistics.hub_server.infrastructure.config;

import chill_logistics.hub_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, HubRouteAfterCreateV1> hubRouteAfterCreateProducerFactory() {

        Map<String, Object> props = new HashMap<>();

        // Kafka Broker
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 직렬화 설정
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        // 필요시 기타 옵션 (acks, retries 등) 추가 가능

        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<String, HubRouteAfterCreateV1> hubRouteAfterCreateKafkaTemplate() {
        return new KafkaTemplate<>(hubRouteAfterCreateProducerFactory());
    }
}

🔸 hub-server: (infrastructure.kafka.dto) HubRouteAfterCreateV1

package chill_logistics.hub_server.infrastructure.kafka.dto;

import java.time.LocalDateTime;
import java.util.UUID;

public record HubRouteAfterCreateV1(
    UUID orderId,
    UUID startHubId,
    String startHubName,
    String startHubFullAddress,
    UUID endHubId,
    String endHubName,
    String endHubFullAddress,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt,
    Integer expectedDeliveryDuration
    ) {}

🔸 hub-server: HubRouteAfterCreteProducer

package chill_logistics.hub_server.infrastructure.kafka;

import chill_logistics.hub_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class HubRouteAfterCreateProducer {

    private final KafkaTemplate<String, HubRouteAfterCreateV1> hubRouteAfterCreateKafkaTemplate;

    @Value("${app.kafka.topic.hub-route-after-create}")
    private String hubRouteAfterCreateTopic;

    /* [허브 경로 생성 후 Kafka로 HubRouteAfterCreate 이벤트 발행]
     */
    public void sendHubRouteAfterCreate(HubRouteAfterCreateV1 message) {

        String key = message.orderId().toString();

        log.info("[Kafka] HubRouteAfterCreate 메시지 발행, topic={}, key={}, message={}",
            hubRouteAfterCreateTopic, key, message);

        hubRouteAfterCreateKafkaTemplate
            .send(hubRouteAfterCreateTopic, key, message)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("[Kafka] HubRouteAfterCreate 메시지 전송 실패, key={}", key, ex);
                } else {
                    log.info("[Kafka] HubRouteAfterCreate 메시지 전송 성공, orderId={}, offset={}",
                        key, result.getRecordMetadata().offset());
                }
            });
    }
}

🔸 hub-server: HubService에서 사용 예시

package chill_logistics.hub_server.application;

import chill_logistics.hub_server.domain.entity.Hub;
import chill_logistics.hub_server.domain.repository.HubRepository;
import chill_logistics.hub_server.infrastructure.kafka.HubRouteAfterCreateProducer;
import chill_logistics.hub_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class HubService {

    private final HubRepository hubRepository;
    private final HubRouteAfterCreateProducer hubRouteAfterCreateProducer;

    @Transactional
    public calculateExpectedDuration() {

        // 1. 예상 소요 시간 계산

        // 2. 저장 (임의로 hub라고 명칭했습니다)
        hubRepository.save(hub);

        // 3. Kafka 메시지 생성
        HubRoutefterCreateV1 message = new HubRoutefterCreateV1(
            hub.getOrderId(),
            hub.getStartHubId(),
            hub.getStartHubName(),
            hub.getStartHubFullAddress(),
            hub.getEndHubId(),
            hub.getEndHubName(),
            hub.getEndHubFullAddress(),
            hub.getReceiverFirmId(),
            hub.getReceiverFirmFullAddress(),
            hub.getReceiverFirmOwnerName(),
            hub.getRequestNote(),
            hub.getProductName(),
            hub.getProductQuantity(),
            hub.getOrderCreatedAt(),
            hub.getExpectedDuration(),
        );

        // 4. Kafka 메시지 발행
        hubRouteAfterCreateProducer.sendHubRouteAfterCreate(message);

        log.info("예상 소요 시간 계산 생성 완료 + Kafka 메시지 발행 완료. orderId={}", hub.getOrderId());

        return hub.getOrderId();
    }
}

🔹 delivery-server (Consumer)

🔸 delivery-server: build.gradle

implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'

🔸 delivery-server: application.yml

spring:
  application:
    name: delivery-server

  kafka:
    bootstrap-servers: localhost:9092

🔸 delivery-server: (infrastructure.config) KafkaConsumerConfig

package chill_logistics.delivery_server.infrastructure.config;

import chill_logistics.delivery_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, HubRouteAfterCreateV1> hubConsumerFactory() {

        // JSON → HubRouteAfterCreateV1 역직렬화를 위한 Deserializer
        JsonDeserializer<HubRouteAfterCreateV1> deserializer =
            new JsonDeserializer<>(HubRouteAfterCreateV1.class, false);

        // Kafka 메시지 역직렬화 시 허용할 패키지를 명시적으로 지정
        deserializer.addTrustedPackages(
            "chill_logistics.delivery_server.infrastructure.kafka.dto"
        );

        // Kafka Consumer 설정 값
        Map<String, Object> properties = new HashMap<>();

        // Kafka Broker 주소 (Docker Compose에서 기본적으로 localhost:9092로 띄움)
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Consumer Group ID (같은 Group으로 묶인 Consumer들은 같은 메시지를 중복 처리하지 않음)
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "delivery-server-group");

        // 메시지 Key 역직렬화 방식
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 메시지 Value 역직렬화 방식 (JsonDeserializer 사용)
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        // 위 설정값 + Key/Value Deserializer를 기반으로 실제 Consumer 인스턴스를 만들어 KafkaListener에 제공
        return new DefaultKafkaConsumerFactory<>(
            properties,
            new StringDeserializer(),   // Key Deserializer (String)
            deserializer                // Value Deserializer (HubRouteAfterCreateV1)
        );
    }

    // @KafkaListener가 사용할 Listener 컨테이너 생성 (ConsumerFactory를 주입하여 실제 Kafka Consumer 동작을 구성)
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, HubRouteAfterCreateV1>
    hubKafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, HubRouteAfterCreateV1> factory =
            new ConcurrentKafkaListenerContainerFactory<>();

        // ConsumerFactory 설정
        factory.setConsumerFactory(hubConsumerFactory());

        return factory;
    }
}

🔸 delivery-server: (infrastructure.kafka.dto) HubRouteAfterCreateV1

package chill_logistics.delivery_server.infrastructure.kafka.dto;

import chill_logistics.delivery_server.application.dto.command.HubRouteAfterCommandV1;
import java.time.LocalDateTime;
import java.util.UUID;

public record HubRouteAfterCreateV1(
    UUID orderId,
    UUID startHubId,
    String startHubName,
    String startHubFullAddress,
    UUID endHubId,
    String endHubName,
    String endHubFullAddress,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt,
    Integer expectedDeliveryDuration) {

    // application 계층의 command 변환 메서드
    public HubRouteAfterCommandV1 toCommand() {
        return new HubRouteAfterCommandV1(
            orderId(),
            startHubId(),
            startHubName(),
            startHubFullAddress(),
            endHubId(),
            endHubName(),
            endHubFullAddress(),
            receiverFirmId(),
            receiverFirmFullAddress(),
            receiverFirmOwnerName(),
            requestNote(),
            productName(),
            productQuantity(),
            orderCreatedAt(),
            expectedDeliveryDuration()
        );
    }
}

🔸 delivery-server: (application.dto) HubRouteAfterCommandV1

package chill_logistics.delivery_server.application.dto.command;

import java.time.LocalDateTime;
import java.util.UUID;

public record HubRouteAfterCommandV1(
    UUID orderId,
    UUID startHubId,
    String startHubName,
    String startHubFullAddress,
    UUID endHubId,
    String endHubName,
    String endHubFullAddress,
    UUID receiverFirmId,
    String receiverFirmFullAddress,
    String receiverFirmOwnerName,
    String requestNote,
    String productName,
    int productQuantity,
    LocalDateTime orderCreatedAt,
    Integer expectedDeliveryDuration
) {}

🔸 delivery-server: HubRouteAfterCreateListener

package chill_logistics.delivery_server.infrastructure.kafka;

import chill_logistics.delivery_server.application.DeliveryCommandService;
import chill_logistics.delivery_server.application.dto.command.HubRouteAfterCommandV1;
import chill_logistics.delivery_server.infrastructure.kafka.dto.HubRouteAfterCreateV1;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class HubRouteAfterCreateListener {

    private final DeliveryCommandService deliveryCommandService;

    @KafkaListener(
        topics = "hub-route-after-create",
        containerFactory = "hubKafkaListenerContainerFactory"
    )
    public void listen(HubRouteAfterCreateV1 message) {

        log.info("Kafka 메시지 수신: {}", message);

        // 허브/업체 배송 담당자 배정 (임시 스텁)
        UUID hubDeliveryPersonId = assignHubDeliveryPerson(message);
        UUID firmDeliveryPersonId = assignFirmDeliveryPerson(message);

        HubRouteAfterCommandV1 command = message.toCommand();

        deliveryCommandService.createDelivery(command, hubDeliveryPersonId, firmDeliveryPersonId);
    }

    // 허브 배송 담당자 배정 (임시 버전 - 이후 배정 로직으로 교체)
    private UUID assignHubDeliveryPerson(HubRouteAfterCreateV1 message) {

        return UUID.fromString("00000000-0000-0000-0000-000000000001");
    }

    // 업체 배송 담당자 배정 (임시 버전 - 이후 배정 로직으로 교체)
    private UUID assignFirmDeliveryPerson(HubRouteAfterCreateV1 message) {

        return UUID.fromString("00000000-0000-0000-0000-000000000002"); // 임시값
    }
}

Kafka code skeleton 참고 링크

DDD 구조의 MSA 프로젝트에서는 JPA의 Pageable 객체를 사용하면 안 좋다고 생각되어 (DDD 규칙에 어긋나기 때문), 커스터마이징 한 Page 객체를 만들었다.

🔹 왜 사용하면 안 되는가?

결론부터 말하면:

“MSA + DDD 구조라서 Page/Pageable을 절대 쓰면 안 된다” 같은 건 없다. 다만, “어디까지 써도 되는지(레이어 경계)” 가 중요하다.

• ✅ 인프라 레이어(Repository 구현체) 에서는 Page/Pageable 마음껏 써도 됨
• ⚠️ 도메인/애플리케이션/공용 인터페이스까지 Page/Pageable이 새어 나오면 좀 곤란해짐
• 그래서 “내 페이징 모델(CustomPage, CustomPageRequest)”을 바깥 계약으로 쓰고, 안쪽에서만 Page/Pageable로 매핑하는 패턴을 많이 씀

🔸 DDD/Hexagonal 관점에서의 문제점

1) 도메인 계층이 Spring Data에 직접 의존하게 됨

DDD + 헥사고날(ports & adapters)에서 보통 구조를 이렇게 잡는다:

• domain/application 레이어: HubRepository 같은 포트 인터페이스
• infra 레이어: SpringDataHubRepository extends JpaRepository<Hub, UUID> 같은 어댑터 구현체

이때 도메인 레이어의 포트 인터페이스에 Page/Pageable이 등장하면:

// (안 좋은 예) domain 레이어에 있는 인터페이스
public interface HubRepository {
    Page<Hub> searchByHubName(String hubName, Pageable pageable);
}

• 도메인 코드가 Spring Data JPA에 직접 의존하게 됨
• 나중에 저장소 기술을 바꾸고 싶을 때(몽고, 엘라스틱, R2DBC 등) → 인터페이스 자체를 갈아엎어야 함
• “도메인은 기술에 독립적이어야 한다”는 DDD/헥사고날 철학이 깨짐

그래서 보통은 이렇게 분리한다:

// domain/application 레이어: 기술 중립 포트
public interface HubRepository {

    MyPage<Hub> searchByHubName(String hubName, MyPageRequest pageRequest);
}

// infra 레이어: Spring Data JPA 어댑터
public interface HubJpaRepository extends JpaRepository<Hub, UUID> {

    Page<Hub> findByHubNameAndDeletedAtIsNull(String hubName, Pageable pageable);
}

// infra 레이어: 포트 구현체
@Repository
@RequiredArgsConstructor
public class HubRepositoryImpl implements HubRepository {

    private final HubJpaRepository hubJpaRepository;

    @Override
    public MyPage<Hub> searchByHubName(String hubName, MyPageRequest pageRequest) {
        Pageable pageable = PageRequest.of(pageRequest.page(), pageRequest.size());
        Page<Hub> page = hubJpaRepository.findByHubNameAndDeletedAtIsNull(hubName, pageable);

        return new MyPage<>(
            page.getContent(),
            page.getNumber(),
            page.getSize(),
            page.getTotalElements()
        );
    }
}

• 도메인/애플리케이션 레이어는 MyPage/MyPageRequest만 알면 됨
• Page/Pageable은 infra 안에서만 사용 → 기술 교체/테스트 분리가 쉬워짐

🔸 MSA 관점에서의 문제점

1) 서비스 간 계약(API, 공용 모듈)에 Spring Data 타입이 새어 나옴

MSA에서는 보통:

• REST API 응답 DTO
• 공용 라이브러리(예: lib-web, lib-pagination)
• 다른 서비스와의 계약(Swagger/OpenAPI)

이런 데서 Spring Data 타입이 바로 노출되는 것을 피하는 편이 좋다.

예를 들어, 이런 응답은 바람직하지 않음:

// (안 좋은 예) Controller에서 Page를 그대로 응답으로 내려버리는 경우
@GetMapping("/hubs")
public Page<HubResponse> searchHubs(...) { ... }

이렇게 되면:

• 외부(클라이언트, 다른 서비스)가 Spring Data의 Page 구조에 종속
• 페이지 인덱스가 0-based인지, 필드명이 뭐인지까지 바깥 계약이 되어버림
• 나중에 페이지 규칙을 바꾸거나 라이브러리 교체하기 어려움

그래서 보통은:

• 내가 정의한 페이징 응답 DTO (HubPageResponseV1)로 감싸고
• 그 안의 content, page, size, totalElements, totalPages 등을 명시적으로 설계

public record HubPageResponseV1(
    List<HubSummaryResponseV1> content,
    int page,
    int size,
    long totalElements,
    int totalPages,
    boolean first,
    boolean last
) {}

• 이렇게 해두면 나중에 JPA → Mongo, JPA → MyBatis로 바꿔도 Controller 바깥 계약은 그대로 유지 가능

🔹 그렇다고 Page/Pageable이 나쁜 건 아니다

어디서 쓰면 좋은지 정리하면:

🔸 이런 곳에서는 적극 사용해도 됨

• infra 레이어(Spring Data JPA repository) 내부

  • JpaRepository 상속, Page<엔티티> find... 메서드
  • PageRequest.of(page, size, Sort...) 로 정렬 포함 페이징

• 테스트 코드, 샘플 코드에서 빠르게 페이징 붙일 때

🔸 이런 곳에선 피하는 게 좋음

• domain/application 레이어 인터페이스 (포트)

  • Page/Pageable이 올라오면 도메인이 Spring Data에 물림

• 서비스 간 API 계약(Controller 응답, 공용 DTO)

  • 프론트/다른 서비스가 Spring Data 타입 구조에 종속됨

• 공용 모듈 (lib-*)

  • 여러 서비스가 함께 쓰는 모듈에 Spring Data 타입을 넣어버리면 그 모듈을 쓰는 모든 서비스가 Spring Data를 강제 당함

🔹 그래서 “MSA + DDD”에서 권장하는 타협안

1. 도메인/애플리케이션/공용에서는

   • CustomPage, CustomPageRequest 같은 기술 중립 페이징 모델 사용

2. infra에서는

   • Spring Data JPA의 Page/Pageable 마음껏 사용
   • 그리고 거기서 CustomPage로 변환해서 위로 올려줌

3. Controller/REST 응답에서는

   • CustomPage 기반 DTO (PageResponseV1)만 노출

이렇게 하면:

• DDD/헥사고날 철학 유지 (도메인은 기술에 독립)
• MSA 간 API 계약도 깔끔하게 유지
• 하지만 내부 구현에서는 Page/Pageable의 생산성을 그대로 가져감

🔹 요약

• “MSA + DDD라서 Page/Pageable 자체가 나쁘다” → ❌ (틀린 말)
• “도메인/애플리케이션/API 계약에 Page/Pageable을 직접 노출하는 건 피하는 게 좋다” → ✅
• 그래서 infra에서만 JPA Page/Pageable 쓰고, 바깥으로는 내가 정의한 MyPage/DTO로 감싸는 패턴이 제일 깔끔하다.

Custom 페이지네이션 구현

🔹 lib (공통 모듈)

분리된 공통 모듈에 커스텀 페이지 구현

🔸 CustomPageRequest

package lib.pagination;

public record CustomPageRequest(
    int page,
    int size) {

    public CustomPageRequest {

        if (page < 0) {
            throw new IllegalArgumentException("페이지는 반드시 0 이상이어야 합니다. page=" + page);
        }

        if (size <= 0) {
            throw new IllegalArgumentException("사이즈는 반드시 0보다 커야 합니다. size=" + size);
        }
    }

    public int offset() {
        return page * size;
    }

    // null 및 이상값 방로용 헬퍼 메서드
    public static CustomPageRequest of(Integer page, Integer size, int defaultPage, int defaultSize) {

        int p = (page == null || page < 0) ? defaultPage : page;
        int s = (size == null || size <= 0) ? defaultSize : size;

        return new CustomPageRequest(p, s);
    }
}

🔸 CustomPageResult

package lib.pagination;

import java.util.List;
import lombok.Getter;

@Getter
public class CustomPageResult<T> {

    private final List<T> itemList;     // 현재 페이지 데이터
    private final int page;             // 현재 페이지 번호
    private final int size;             // 페이지 크기
    private final long totalCount;      // 전체 데이터 개수 (Slice면 -1 허용)
    private final boolean hasNext;      // 다음 페이지 존재 여부

    private CustomPageResult(List<T> itemList, int page, int size, long totalCount, boolean hasNext) {

        this.itemList = itemList;
        this.page = page;
        this.size = size;
        this.totalCount = totalCount;
        this.hasNext = hasNext;
    }

    // totalCount를 아는 일반적인 페이징용
    public static <T> CustomPageResult<T> of(List<T> itemList, int page, int size, long totalCount) {

        boolean hasNext = ((long) page + 1) * size < totalCount;

        return new CustomPageResult<>(itemList, page, size, totalCount, hasNext);
    }

    // totalCount를 모르는 Slice 기반 페이징용
    public static <T> CustomPageResult<T> sliceOf(List<T> itemList, int page, int size, boolean hasNext) {
        return new CustomPageResult<>(itemList, page, size, -1, hasNext);
    }
}

🔹 다른 모듈에 사용 예시

🔸 HubDeliveryRepository

package chill_logistics.delivery_server.domain.repository;

import chill_logistics.delivery_server.domain.entity.HubDelivery;
import lib.pagination.CustomPageRequest;
import lib.pagination.CustomPageResult;

public interface HubDeliveryRepository {

    CustomPageResult<HubDelivery> searchByStartHubName(String startHubName, CustomPageRequest customPageRequest);
}

🔸HubDeliveryRepositoryAdapter

package chill_logistics.delivery_server.infrastructure.repository;

import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import lib.pagination.CustomPageRequest;
import lib.pagination.CustomPageResult;
import lombok.RequiredArgsConstructor;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;

@RequiredArgsConstructor
public class HubDeliveryRepositoryAdapter implements HubDeliveryRepository {

    private final JpaHubDeliveryRepository jpaHubDeliveryRepository;

    @Override
    public CustomPageResult<HubDelivery> searchByStartHubName(
        String startHubName,
        CustomPageRequest customPageRequest) {

        PageRequest pageable = PageRequest.of(customPageRequest.page(), customPageRequest.size());

        Page<HubDelivery> page;

        // 검색어 없으면 전체 조회, 있으면 조건 검색
        if (startHubName == null || startHubName.isBlank()) {
            page = jpaHubDeliveryRepository.findByDeletedAtIsNull(pageable);
        } else {
            page = jpaHubDeliveryRepository.findByStartHubNameAndDeletedAtIsNull(startHubName, pageable);
        }

        return CustomPageResult.of(
            page.getContent(),
            page.getNumber(),
            page.getSize(),
            page.getTotalElements()
        );
    }
}

🔸JpaHubDeliveryRepository

package chill_logistics.delivery_server.infrastructure.repository;

import chill_logistics.delivery_server.domain.entity.HubDelivery;
import java.util.UUID;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;

public interface JpaHubDeliveryRepository extends JpaRepository<HubDelivery, UUID> {

    Page<HubDelivery> findByDeletedAtIsNull(Pageable pageable);

    Page<HubDelivery> findByStartHubNameAndDeletedAtIsNull(String hubName, Pageable pageable);
}

🔸DeliveryQueryService

package chill_logistics.delivery_server.application;

import chill_logistics.delivery_server.application.dto.query.HubDeliveryInfoResponseV1;
import chill_logistics.delivery_server.application.dto.query.HubDeliverySummaryResponseV1;
import chill_logistics.delivery_server.domain.entity.HubDelivery;
import chill_logistics.delivery_server.domain.repository.FirmDeliveryRepository;
import chill_logistics.delivery_server.domain.repository.HubDeliveryRepository;
import chill_logistics.delivery_server.presentation.ErrorCode;
import chill_logistics.delivery_server.presentation.dto.response.HubDeliveryPageResponseV1;
import java.util.List;
import java.util.UUID;
import lib.pagination.CustomPageRequest;
import lib.pagination.CustomPageResult;
import lib.web.error.BusinessException;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class DeliveryQueryService {

    private final HubDeliveryRepository hubDeliveryRepository;
    private final FirmDeliveryRepository firmDeliveryRepository;

    /* [허브배송 단건 조회]
     */
    public HubDeliveryInfoResponseV1 getHubDelivery(UUID hubDeliveryId) {

        HubDelivery hubDelivery = hubDeliveryRepository.findById(hubDeliveryId)
            .orElseThrow(() -> new BusinessException(ErrorCode.HUB_DELIVERY_NOT_FOUND));

        if (!(hubDelivery.getDeletedAt() == null)) {
            throw new BusinessException(ErrorCode.DELIVERY_HAS_BEEN_DELETED);
        }

        return HubDeliveryInfoResponseV1.from(hubDelivery);
    }

    /* [허브배송 검색 조회]
     * 검색 기준: startHubName
     * 검색어 없으면 전체 목록 조회, 있으면 조건 검색 결과 반환
     */
    public HubDeliveryPageResponseV1 searchHubDeliveryByHubName(String hubName, int page, int size) {

        CustomPageRequest pageRequest = new CustomPageRequest(page, size);

        // 페이징 된 엔티티 조회
        CustomPageResult<HubDelivery> entityPage = hubDeliveryRepository.searchByStartHubName(hubName, pageRequest);

        // 엔티티 → DTO로 변환
        List<HubDeliverySummaryResponseV1> dataSummaryList = entityPage.getItemList().stream()
            .map(HubDeliverySummaryResponseV1::from)
            .toList();

        return HubDeliveryPageResponseV1.of(entityPage, dataSummaryList);
    }
}

🔸DeliveryController

package chill_logistics.delivery_server.presentation;

import chill_logistics.delivery_server.application.DeliveryQueryService;
import chill_logistics.delivery_server.application.dto.query.HubDeliveryInfoResponseV1;
import chill_logistics.delivery_server.presentation.dto.response.HubDeliveryPageResponseV1;
import java.util.UUID;
import lib.entity.BaseStatus;
import lib.web.response.BaseResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequiredArgsConstructor
@RequestMapping("/v1")
public class DeliveryController {

    private final DeliveryQueryService deliveryQueryService;

    /**
     * [허브배송 단건 조회]
     *
     * @param hubDeliveryId 조회하고자 하는 허브배송의 UUID
     * @return 허브배송 상세 정보
     */
    @GetMapping("/hub-deliveries/{hubDeliveryId}")
    @ResponseStatus(HttpStatus.OK)
    public BaseResponse<HubDeliveryInfoResponseV1> getHubDelivery(
        @PathVariable("hubDeliveryId") UUID hubDeliveryId) {

        HubDeliveryInfoResponseV1 response = deliveryQueryService.getHubDelivery(hubDeliveryId);

        return BaseResponse.ok(response, BaseStatus.OK);
    }

    /**
     * [허브배송 검색 조회]
     *
     * @param startHubName 허브배송에서 검색하고자 하는 허브명
     * @param page         조회할 페이지 번호 (0부터 시작)
     * @param size         페이지 당 조회할 데이터 개수
     * @return 허브배송 요약 정보 목록 + 페이징 정보
     */
    @GetMapping("/hub-deliveries")
    @ResponseStatus(HttpStatus.OK)
    public BaseResponse<HubDeliveryPageResponseV1> searchHubDeliveries(
        @RequestParam(required = false) String startHubName,
        @RequestParam(defaultValue = "0") int page,
        @RequestParam(defaultValue = "20") int size) {

        HubDeliveryPageResponseV1 response = deliveryQueryService.searchHubDeliveryByHubName(
            startHubName, page, size);

        return BaseResponse.ok(response, BaseStatus.OK);
    }
}