Kafka 마스터 — Spring Cloud Stream·Reactive Kafka

이 글은 카프카 마스터 노트 시리즈의 아홉 번째 편입니다. 기초편(1~8편)에서 Kafka·Spring Kafka를 다졌다면, 이번엔 그 위 한 단계 추상화 — Spring Cloud Stream과 Saga 패턴.

이 시리즈 15편은 SCS 기초·StreamBridge·Fan-Out/In·Tips·Saga 코레오그래피·Saga 오케스트레이터·Transactional Outbox까지. 1편의 목표 — SCS의 Binder/Binding 개념과 Reactive Kafka 통합.

처음 Spring Cloud Stream이 어렵게 느껴지는 이유

처음 이 단원이 어렵게 느껴지는 이유는 두 가지예요. 첫째, Binder·Binding 두 단어가 비슷합니다. 누가 누구의 부분집합인가? 둘째, 함수형 인터페이스(Supplier·Consumer·Function)로 빈을 등록하는 방식이 익숙하지 않습니다. @KafkaListener는 명확한데 SCS는 왜 이런 구조?

해결법은 한 가지예요. "Binder = 구현체 / Binding = 채널" 비유로 묶는 것. Binder = 운송 회사(카프카 회사·RabbitMQ 회사), Binding = 그 회사가 만들어 준 실제 운송 라인. 함수형 빈 = 그 라인의 입출구 역할. 이 그림이 잡히면 모든 동작이 따라옵니다.

Spring Cloud Stream — 메시징 추상화

왜 SCS인가

Spring Kafka:

@KafkaListener(topics = "order-events")
public void onOrder(OrderEvent event) { ... }

SCS:

@Bean
public Consumer<OrderEvent> orderProcessor() {
    return event -> { ... };
}

SCS의 가치 — 메시징 플랫폼 비종속. 같은 코드가 카프카로도 RabbitMQ로도 동작.

# Kafka 설정
spring.cloud.stream.binders.kafka.type: kafka

# RabbitMQ로 전환만 — 코드 변경 X
spring.cloud.stream.binders.rabbit.type: rabbit

여기서 정말 중요한 시험 함정 — SCS는 Spring Data JPA와 비슷한 추상화. JPA가 DB 추상화처럼, SCS는 메시징 추상화. 코드와 인프라 분리.

Binder vs Binding

Binder:  메시징 시스템과의 연결 구현체
         (Kafka Binder, RabbitMQ Binder, ...)
Binding: Binder가 생성하는 실제 메시지 채널
         (input·output 채널)

Application
   ↓
[Binding: orderEvents-out-0]   ← 이게 채널
   ↓
[Binder: kafka]               ← 이게 구현
   ↓
[Kafka Broker]

함수형 빈 3종 — Supplier/Consumer/Function

SCS는 Spring Cloud Function 위에 올라감. 함수형 인터페이스 빈을 등록하면 자동 바인딩.

1. Supplier — 발행 (메시지 만듦)

@Bean
public Supplier<OrderEvent> orderSupplier() {
    return () -> new OrderEvent(...);
}

자동 출력 채널 — orderSupplier-out-0. 주기적으로 호출되며 메시지 발행.

2. Consumer — 소비 (메시지 받음)

@Bean
public Consumer<OrderEvent> orderProcessor() {
    return event -> {
        log.info("Processing: {}", event);
        // 처리
    };
}

자동 입력 채널 — orderProcessor-in-0.

3. Function — 변환 (받고 발행)

@Bean
public Function<OrderEvent, PaymentEvent> orderToPayment() {
    return order -> new PaymentEvent(order.getOrderId(), order.getAmount());
}

입력 + 출력 — orderToPayment-in-0, orderToPayment-out-0.

여기서 정말 중요한 시험 함정 — 빈 이름이 바인딩 이름의 prefix. 함수형 빈 이름이 변경되면 바인딩 이름도 변경. 신중히 명명.

Reactive 변형 — Flux/Mono

@Bean
public Function<Flux<OrderEvent>, Flux<PaymentEvent>> processOrders() {
    return orders -> orders
        .flatMap(this::processAsync, 8)
        .map(this::toPayment);
}

Reactive 형태가 권장. 백프레셔·논블로킹 자연스러움.

바인딩 명명 규칙

<빈이름>-<in|out>-<인덱스>

예시:

orderToPayment-in-0    ← Function 첫 입력
orderToPayment-out-0   ← Function 첫 출력
processOrder-in-0      ← Consumer 입력
notifySupplier-out-0   ← Supplier 출력

여기서 시험 함정이 하나 있어요. 인덱스는 0부터. 다중 입출력 시 (Tuples) 1, 2 증가.

application.yaml 설정

spring:
  cloud:
    function:
      definition: orderToPayment;processOrder    # 활성화할 빈

    stream:
      bindings:
        orderToPayment-in-0:
          destination: order-events              # 카프카 토픽
          group: payment-service                 # consumer group
          content-type: application/json
        orderToPayment-out-0:
          destination: payment-events
        processOrder-in-0:
          destination: notification-events
          group: notification-service

      # Kafka Binder 전용
      kafka:
        binder:
          brokers: localhost:9092
          configuration:
            auto.offset.reset: earliest
        bindings:
          orderToPayment-in-0:
            consumer:
              ack-mode: MANUAL

핵심 — function.definition에 빈 이름 명시 (세미콜론 구분).

ReceiverOptionCustomizer — 프로그래밍 방식 설정

YAML로 못 표현하는 세밀 설정:

@Bean
public ReceiverOptionsCustomizer<String, OrderEvent> customCustomizer() {
    return (name, options) -> options
        .commitInterval(Duration.ofSeconds(5))
        .commitBatchSize(100)
        .pollTimeout(Duration.ofMillis(500));
}

name은 binding 이름. 이름 분기로 binding별 다른 설정 가능.

SenderOptionsCustomizer도 비슷.

SenderResult — 발행 결과 처리

@Bean
public Function<Flux<OrderEvent>, Flux<SenderResult<Object>>> processWithResult() {
    return orders -> orders
        .map(this::toRecord)
        .flatMap(record -> kafkaSender.send(Mono.just(record)));
}

발행 성공·실패를 명시적으로 받음. 운영 모니터링·알람에 유용.

통합 테스트 패턴

Embedded Kafka

@EmbeddedKafka(partitions = 1, topics = {"order-events", "payment-events"})
@SpringBootTest
class SCSTest {
    @Autowired
    private TestPublisher<OrderEvent> publisher;

    @Test
    void test() {
        // ...
    }
}

StepVerifier + Sinks

@Bean
public Sinks.Many<OrderEvent> testSink() {
    return Sinks.many().multicast().onBackpressureBuffer();
}

@Bean
public Supplier<Flux<OrderEvent>> testSupplier(Sinks.Many<OrderEvent> sink) {
    return sink::asFlux;
}

// 테스트
testSink.tryEmitNext(new OrderEvent(...));

auto.offset.reset=earliest 설정

spring.cloud.stream.kafka.binder.configuration.auto.offset.reset: earliest

테스트에서 메시지 못 받는 흔한 함정 회피.

SCS의 한계

여기서 정말 중요한 시험 함정 — Reactive Kafka Binder는 Kafka 트랜잭션 미지원. 7편(1~8편)에서 본 transactional.id 없음. 정확히 한 번이 필요하면 Outbox Pattern (이 시리즈 15편).

전통 Spring Kafka vs SCS

측면	Spring Kafka	Spring Cloud Stream
추상화	Kafka에 특화	메시징 일반
설정	Kafka 옵션 직접	YAML 추상
Listener	@KafkaListener	Function/Consumer 빈
트랜잭션	지원	Reactive Binder 미지원
다중 메시징	X	O (Kafka·Rabbit 등)

선택:

• 카프카 전용 + 트랜잭션 → Spring Kafka
• WebFlux + 메시징 추상화 → Reactive SCS

시험 직전 한 번 더 — 자주 헷갈리는 함정 모음

여기까지가 1편의 핵심입니다. 시험 직전 또는 실무에서 헷갈릴 때 다시 펼쳐 볼 수 있게 압축 노트로 마무리할게요.

• Spring Cloud Stream = 메시징 추상화 (JPA가 DB 추상화처럼)
• 같은 코드 = 카프카·RabbitMQ 둘 다
• Binder = 메시징 시스템 구현체 (Kafka Binder)
• Binding = 실제 채널 (input·output)
• 함수형 빈 3종 — Supplier (발행) / Consumer (소비) / Function (변환)
• Spring Cloud Function 위에 동작
• Reactive 변형 — Flux/Mono 기반 (권장)
• 바인딩 명명 — <빈이름>-<in|out>-<인덱스>
• 인덱스 0부터
• function.definition 에 빈 이름 명시 (세미콜론)
• bindings에 destination(토픽)·group·content-type
• Kafka Binder 전용 속성 — binder.configuration, binder.consumer, binder.producer
• ReceiverOptionCustomizer = 프로그래밍 방식 세밀 설정
• name 분기로 binding별 차등
• SenderResult = 발행 성공·실패 명시적
• 테스트 — Embedded Kafka + StepVerifier + Sinks
• auto.offset.reset=earliest 명시 (메시지 못 받는 흔한 함정)
• Reactive Kafka Binder는 Kafka 트랜잭션 미지원 → Outbox Pattern 필요
• Spring Kafka vs SCS — 카프카 전용+TX vs 메시징 추상

시리즈 다른 편

• 1편 — EDA·Kafka 기초·KRaft
• 2편 — Topic·Partition·Offset
• 3편 — Producer·Consumer 동작
• 4편 — Consumer Group·리밸런싱
• 5편 — Reactor Kafka
• 6편 — Cluster·HA·Best Practices
• 7편 — 배치·에러·트랜잭션
• 8편 — Spring Kafka·테스트·보안
• 9편 — Spring Cloud Stream 기초 (현재 글)
• 10편 — StreamBridge 동적 라우팅
• 11편 — Fan-Out / Fan-In
• 12편 — SCS Tips & Tricks
• 13편 — Saga 코레오그래피
• 14편 — Saga 오케스트레이터
• 15편 — Transactional Outbox

공식 문서: Spring Cloud Stream Reference 에서 더 깊이.

다음 글(2편)에서는 StreamBridge — 정적 바인딩의 한계를 넘는 동적 라우팅과 콘텐츠 기반 라우팅까지 풀어 갑니다.