예제 2 · Spring Boot Producer/Consumer 기본

학습 목표

spring.kafka.* 프로퍼티가 어떤 Kafka 설정으로 매핑되는지, 전용 키가 없는 설정은 어디에 넣는지 알 수 있습니다.
KafkaTemplate.send()가 반환하는 CompletableFuture를 버리지 않고 다루는 방법을 씁니다.
ack-mode 7종의 차이를 알고, 수동 ack(MANUAL_IMMEDIATE)를 코드로 구현할 수 있습니다.
concurrency와 파티션 수의 관계를 설명할 수 있습니다.

시나리오

사내 알림 서비스가 여러 도메인 서비스로부터 알림 요청을 받습니다. 기존에는 HTTP 동기 호출이라 알림 서비스가 느려지면 호출한 서비스의 응답 시간까지 함께 늘어났습니다.

이를 Kafka로 분리합니다. 도메인 서비스는 notifications 토픽에 요청을 넣고 즉시 반환하고, 알림 서비스는 그 토픽을 소비해 실제 발송을 합니다. 첫 단계 목표는 가장 단순하지만 유실이 없는 구조를 만드는 것입니다.

아키텍처

하나의 Spring Boot 애플리케이션이 프로듀서와 컨슈머를 모두 갖습니다 (실제로는 서비스가 분리되지만, 학습을 위해 한 프로세스에 둡니다). REST 엔드포인트가 프로듀서를 호출하고, @KafkaListener가 소비합니다.

Spring for Apache Kafka의 주요 구성 요소와 대응하는 Kafka 클라이언트
Spring 컴포넌트	감싸는 Kafka 객체	역할
`ProducerFactory`	`KafkaProducer` 생성	프로듀서 설정을 보관하고 인스턴스를 만듭니다. 기본적으로 하나를 공유합니다
`KafkaTemplate`	`KafkaProducer.send()`	`CompletableFuture<SendResult>`를 반환합니다. 3.0부터 `ListenableFuture`가 아닙니다
`ConsumerFactory`	`KafkaConsumer` 생성	컨슈머 설정을 보관합니다
`ConcurrentMessageListenerContainer`	`KafkaConsumer.poll()` 루프	`concurrency` 개의 컨슈머 스레드를 돌립니다. 각 스레드가 별개의 `KafkaConsumer`입니다
`@KafkaListener`	—	poll로 받은 레코드를 메서드 호출로 변환합니다
`KafkaAdmin` + `NewTopic` 빈	`AdminClient.createTopics()`	기동 시 토픽을 선언적으로 만듭니다 (`spring.kafka.admin.auto-create`, 기본 `true`)

사전 요구사항

검증 환경 (2026-07 기준)
항목	버전	비고
Apache Kafka (브로커)	4.3.1	예제 1의 클러스터
Spring Boot	4.1.0	Maven Central `spring-boot-starter-parent` 최신 정식 버전
Spring for Apache Kafka	4.1.0	Spring Boot 4.1.0이 관리합니다. 버전을 직접 적지 않습니다
kafka-clients	4.3.1로 오버라이드	Spring Boot 4.1.0의 기본 관리 버전은 `4.2.1`입니다
Java	17 이상	Spring Boot 4의 기준선

Spring Boot 3.x와 4.x의 차이

의존성	3.x: `org.springframework.kafka:spring-kafka` 직접 선언 → 4.x: `org.springframework.boot:spring-boot-starter-kafka`
자동 설정 패키지	3.x: `org.springframework.boot.autoconfigure.kafka` → 4.x: `org.springframework.boot.kafka.autoconfigure`
`send()` 반환형	2.x: `ListenableFuture` → 3.0부터 `CompletableFuture` (4.x도 동일)

전체 코드

디렉터리 구조

notification-service/
├── pom.xml
└── src/main/
    ├── java/com/example/notification/
    │   ├── NotificationApplication.java
    │   ├── KafkaTopicConfig.java      # 토픽 선언
    │   ├── NotificationRequest.java   # 이벤트 레코드
    │   ├── NotificationPublisher.java # 프로듀서
    │   ├── NotificationListener.java  # 컨슈머 (수동 ack)
    │   └── NotificationController.java# REST 엔드포인트
    └── resources/
        └── application.yml

notification-service/pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.0</version>
    <relativePath/>
  </parent>

  <groupId>com.example</groupId>
  <artifactId>notification-service</artifactId>
  <version>1.0.0</version>

  <properties>
    <java.version>17</java.version>
    <!-- 브로커(4.3.1)와 클라이언트 버전을 맞춥니다.
         Spring Boot 4.1.0 의 기본 관리 버전은 4.2.1 입니다. -->
    <kafka.version>4.3.1</kafka.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-validation</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

`application.yml` — 프로퍼티가 어디로 가는지

src/main/resources/application.yml

server:
  port: 8080

spring:
  application:
    name: notification-service

  kafka:
    # 초기 연결 이중화. 부하 분산 목적이 아닙니다.
    bootstrap-servers: localhost:29092,localhost:39092,localhost:49092

    # NewTopic 빈으로 선언한 토픽을 기동 시 만듭니다. 기본값 true.
    admin:
      auto-create: true
      # 브로커에 붙지 못해도 애플리케이션 기동을 막지 않습니다(로컬 개발 편의).
      # 프로덕션에서는 true 로 두어 설정 오류를 조기에 드러내는 편이 낫습니다.
      fail-fast: false

    producer:
      # → acks. 4.3 기본값도 all 이지만 의도를 명시합니다.
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # JsonSerializer 는 spring-kafka 가 제공합니다(Jackson 필요).
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      # → batch.size (바이트). 기본값 16384.
      batch-size: 32768
      # → buffer.memory. 기본값 33554432(32MB).
      buffer-memory: 33554432
      # → compression.type. 기본값 none.
      compression-type: lz4
      properties:
        # spring.kafka.producer 아래에 전용 키가 없는 설정은 여기에 씁니다.
        # 프로퍼티 이름은 Kafka 원래 이름을 그대로 씁니다.
        enable.idempotence: true
        # 4.0 에서 기본값이 0 → 5 로 바뀌었습니다. 명시해 의도를 남깁니다.
        linger.ms: 5
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000
        request.timeout.ms: 30000
        max.block.ms: 60000

    consumer:
      group-id: notification-sender
      # 새 그룹일 때만 적용됩니다. 기본값은 latest 입니다.
      # earliest 로 두면 새 그룹이 처음부터 전부 읽습니다.
      auto-offset-reset: earliest
      # → enable.auto.commit. Kafka 기본값은 true 입니다.
      # false 로 두고 컨테이너(ack-mode)가 커밋을 통제하게 합니다.
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      # → max.poll.records. 기본값 500.
      max-poll-records: 100
      # → max.poll.interval.ms. 기본값 300000(5분).
      max-poll-interval: 300s
      # → session.timeout.ms. 기본값 45000. (classic 프로토콜에서만 쓰입니다)
      session-timeout: 45s
      # → heartbeat.interval.ms. 기본값 3000.
      heartbeat-interval: 3s
      properties:
        # JsonDeserializer 가 역직렬화할 대상 패키지를 화이트리스트로 지정합니다.
        # "*" 로 두면 임의 클래스를 역직렬화할 수 있어 위험합니다.
        spring.json.trusted.packages: com.example.notification
        # 상류가 트랜잭션 프로듀서라면 read_committed 가 필요합니다(예제 5).
        # 기본값은 read_uncommitted 입니다.
        isolation.level: read_uncommitted

    listener:
      # MANUAL_IMMEDIATE: 리스너가 Acknowledgment.acknowledge() 를 호출하는 즉시 커밋합니다.
      # 코드가 커밋 시점을 완전히 통제하므로 유실/중복 경계를 직접 설계할 수 있습니다.
      ack-mode: MANUAL_IMMEDIATE
      # 이 인스턴스가 만드는 컨슈머 스레드 수.
      # 파티션 수(3)를 넘기면 남는 스레드는 유휴 상태가 됩니다.
      concurrency: 3
      # 구독 토픽이 없으면 기동 실패 → 토픽명 오타를 조기에 발견합니다.
      missing-topics-fatal: true
      poll-timeout: 3s

app:
  topics:
    notifications: notifications
    notifications-partitions: 3
    notifications-replicas: 3

management:
  endpoints:
    web:
      exposure:
        include: health,metrics,info

logging:
  level:
    com.example.notification: DEBUG

토픽 선언

src/main/java/com/example/notification/KafkaTopicConfig.java

package com.example.notification;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

/**
 * 토픽을 코드로 선언합니다.
 *
 * NewTopic 빈이 있으면 KafkaAdmin 이 기동 시 생성합니다
 * (spring.kafka.admin.auto-create, 기본값 true).
 * 이미 존재하면 만들지 않습니다. 파티션 수가 선언보다 적으면 늘려 줍니다
 * (줄이는 것은 Kafka 자체가 불가능합니다).
 *
 * 자동 생성(auto.create.topics.enable)에 맡기면 파티션 1 / RF 1 이 됩니다.
 */
@Configuration
public class KafkaTopicConfig {

    @Bean
    public NewTopic notificationsTopic(
            @Value("${app.topics.notifications}") String name,
            @Value("${app.topics.notifications-partitions}") int partitions,
            @Value("${app.topics.notifications-replicas}") int replicas) {

        return TopicBuilder.name(name)
                .partitions(partitions)
                .replicas(replicas)
                // acks=all 과 짝을 이루는 내구성 하한. RF=3 에 2 가 표준입니다.
                // 브로커 기본값은 1 이므로 반드시 토픽 레벨로 명시합니다.
                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
                // 알림 요청은 3일이면 충분합니다. 기본값은 7일(604800000)입니다.
                .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(3L * 24 * 60 * 60 * 1000))
                .build();
    }
}

이벤트 레코드

src/main/java/com/example/notification/NotificationRequest.java

package com.example.notification;

import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;

import java.time.Instant;

/**
 * 알림 요청 이벤트.
 *
 * record 를 쓰면 불변이라 스레드 안전하고, JsonSerializer 가 그대로 직렬화합니다.
 * 실무에서는 여기에 스키마(Avro/Protobuf)를 붙이는 편이 안전합니다(예제 7).
 */
public record NotificationRequest(
        @NotBlank String requestId,
        @NotBlank String userId,
        @NotNull Channel channel,
        @NotBlank String message,
        Instant requestedAt) {

    public enum Channel { EMAIL, SMS, PUSH }

    /** 발행 시각을 서버에서 채웁니다. */
    public static NotificationRequest of(String requestId, String userId,
                                         Channel channel, String message) {
        return new NotificationRequest(requestId, userId, channel, message, Instant.now());
    }
}

프로듀서 — `CompletableFuture`를 버리지 않습니다

src/main/java/com/example/notification/NotificationPublisher.java

package com.example.notification;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;

@Service
public class NotificationPublisher {

    private static final Logger log = LoggerFactory.getLogger(NotificationPublisher.class);

    private final KafkaTemplate<String, NotificationRequest> template;
    private final String topic;

    public NotificationPublisher(KafkaTemplate<String, NotificationRequest> template,
                                 @Value("${app.topics.notifications}") String topic) {
        this.template = template;
        this.topic = topic;
    }

    /**
     * 비동기 발행.
     *
     * KafkaTemplate.send() 는 spring-kafka 3.0부터 CompletableFuture 를 반환합니다
     * (그 이전은 ListenableFuture 였습니다).
     * 반환값을 버리면 브로커가 거부해도 아무 일도 일어나지 않은 것처럼 보입니다.
     */
    public CompletableFuture<SendResult<String, NotificationRequest>> publish(
            NotificationRequest request) {

        // 키를 userId 로 둡니다. 같은 사용자의 알림 순서가 보장되고,
        // 파티션 분포도 사용자 수만큼 고르게 퍼집니다.
        // requestId 를 키로 쓰면 순서 보장의 의미가 없어집니다(항상 유일하므로).
        ProducerRecord<String, NotificationRequest> record =
                new ProducerRecord<>(topic, request.userId(), request);

        // 추적 헤더. 컨슈머가 본문을 역직렬화하지 못해도 읽을 수 있습니다.
        record.headers()
              .add("request-id", request.requestId().getBytes(StandardCharsets.UTF_8))
              .add("channel", request.channel().name().getBytes(StandardCharsets.UTF_8));

        CompletableFuture<SendResult<String, NotificationRequest>> future = template.send(record);

        // whenComplete 로 성공/실패를 모두 관측합니다.
        // 이 콜백은 프로듀서의 Sender(I/O) 스레드에서 실행되므로 블로킹 금지입니다.
        future.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("발행 실패 requestId={} userId={}",
                        request.requestId(), request.userId(), ex);
                return;
            }
            var md = result.getRecordMetadata();
            log.debug("발행 성공 requestId={} → {}-{}@{}",
                    request.requestId(), md.topic(), md.partition(), md.offset());
        });

        return future;
    }

    /**
     * 동기 발행 — 호출자가 결과를 반드시 알아야 할 때만 씁니다.
     *
     * join() 은 배치를 무력화하므로 초당 수천 건 경로에는 쓰지 마세요.
     * (레코드마다 브로커 왕복을 기다립니다)
     */
    public SendResult<String, NotificationRequest> publishSync(NotificationRequest request) {
        // join() 은 실패 시 CompletionException 으로 감싸 던집니다.
        return publish(request).join();
    }
}

컨슈머 — 수동 ack

src/main/java/com/example/notification/NotificationListener.java

package com.example.notification;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class NotificationListener {

    private static final Logger log = LoggerFactory.getLogger(NotificationListener.class);

    @KafkaListener(
            topics = "${app.topics.notifications}",
            groupId = "notification-sender")
    public void onNotification(
            @Payload NotificationRequest request,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.debug("수신 {}-{}@{} userId={} channel={}",
                topic, partition, offset, request.userId(), request.channel());

        // 실제 발송 로직. 실패하면 예외를 던집니다.
        // 여기서 try-catch 로 삼키면 에러 핸들러가 동작하지 못합니다(예제 6).
        send(request);

        // ack-mode=MANUAL_IMMEDIATE 이므로 이 호출 즉시 오프셋이 커밋됩니다.
        // 처리 "후" 에 커밋하므로 at-least-once 입니다 —
        // 커밋 전에 죽으면 같은 레코드를 다시 받습니다(중복 가능, 유실 없음).
        // 반대로 처리 "전" 에 ack 하면 at-most-once 가 됩니다(유실 가능).
        ack.acknowledge();
    }

    private void send(NotificationRequest request) {
        // 실제로는 이메일/SMS/푸시 게이트웨이를 호출합니다.
        // 재시도 시 중복 발송을 막으려면 requestId 를 멱등 키로 넘겨야 합니다.
        log.info("발송 channel={} userId={} message={}",
                request.channel(), request.userId(), request.message());
    }
}

REST 엔드포인트

src/main/java/com/example/notification/NotificationController.java

package com.example.notification;

import jakarta.validation.Valid;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletionException;

@RestController
@RequestMapping("/api/notifications")
public class NotificationController {

    private static final Logger log = LoggerFactory.getLogger(NotificationController.class);

    private final NotificationPublisher publisher;

    public NotificationController(NotificationPublisher publisher) {
        this.publisher = publisher;
    }

    public record CreateRequest(String userId,
                                NotificationRequest.Channel channel,
                                String message) {
    }

    /**
     * 비동기 발행 — 202 Accepted 를 즉시 반환합니다.
     * 발행 실패는 로그와 메트릭으로만 드러나므로,
     * 유실이 곤란한 요청이라면 아래 동기 엔드포인트를 쓰세요.
     */
    @PostMapping
    public ResponseEntity<Map<String, String>> create(@Valid @RequestBody CreateRequest body) {
        String requestId = UUID.randomUUID().toString();
        publisher.publish(NotificationRequest.of(
                requestId, body.userId(), body.channel(), body.message()));

        return ResponseEntity.accepted().body(Map.of("requestId", requestId));
    }

    /**
     * 동기 발행 — 브로커 커밋을 확인한 뒤 201 을 반환합니다.
     * 실패하면 503 을 반환해 호출자가 재시도할 수 있게 합니다.
     */
    @PostMapping("/sync")
    public ResponseEntity<Map<String, String>> createSync(@Valid @RequestBody CreateRequest body) {
        String requestId = UUID.randomUUID().toString();
        try {
            var result = publisher.publishSync(NotificationRequest.of(
                    requestId, body.userId(), body.channel(), body.message()));
            var md = result.getRecordMetadata();
            return ResponseEntity.status(HttpStatus.CREATED).body(Map.of(
                    "requestId", requestId,
                    "partition", String.valueOf(md.partition()),
                    "offset", String.valueOf(md.offset())));

        } catch (CompletionException e) {
            // join() 이 감싸 던진 원인을 풀어 로그에 남깁니다.
            log.error("동기 발행 실패 requestId={}", requestId, e.getCause());
            return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
                    .body(Map.of("requestId", requestId, "error", "kafka-unavailable"));
        }
    }
}

src/main/java/com/example/notification/NotificationApplication.java

package com.example.notification;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class NotificationApplication {

    public static void main(String[] args) {
        SpringApplication.run(NotificationApplication.class, args);
    }
}

`ack-mode` 7종

spring-kafka `ContainerProperties.AckMode` — `spring.kafka.listener.ack-mode`로 지정
값	커밋 시점	언제 쓰나
`RECORD`	리스너가 레코드 하나를 성공 처리한 직후	재처리 비용이 큰 경우. 커밋 요청이 가장 많습니다
`BATCH`	poll 배치의 모든 레코드 처리 후 (기본값)	일반적인 선택. 처리량과 안전의 균형
`TIME`	`ack-time` 경과 후	커밋 빈도를 시간으로 제한하고 싶을 때
`COUNT`	`ack-count`건 처리 후	커밋 빈도를 건수로 제한하고 싶을 때
`COUNT_TIME`	둘 중 먼저 도달한 쪽	위 두 개의 조합
`MANUAL`	`acknowledge()` 호출을 모아 두고 배치 끝에 커밋	코드가 대상을 고르되 커밋 요청 수를 줄이고 싶을 때
`MANUAL_IMMEDIATE`	`acknowledge()` 호출 즉시 커밋	커밋 시점을 완전히 통제해야 할 때. 이 예제가 쓰는 값

실행 방법

순서대로 실행

# 0. 예제 1의 클러스터
cd kafka-lab && docker compose ps
cd ..

# 1. 애플리케이션 기동 (토픽은 NewTopic 빈이 만듭니다)
cd notification-service
mvn -q spring-boot:run

# 2. 다른 터미널에서 비동기 발행
curl -s -X POST http://localhost:8080/api/notifications \
  -H 'Content-Type: application/json' \
  -d '{"userId":"U-1001","channel":"EMAIL","message":"주문이 접수되었습니다"}'

# 3. 동기 발행 — 파티션과 오프셋이 응답에 담깁니다
curl -s -X POST http://localhost:8080/api/notifications/sync \
  -H 'Content-Type: application/json' \
  -d '{"userId":"U-1001","channel":"SMS","message":"배송이 시작되었습니다"}'

# 4. 같은 사용자로 10건 — 같은 파티션으로 가는지 확인
for i in $(seq 1 10); do
  curl -s -X POST http://localhost:8080/api/notifications/sync \
    -H 'Content-Type: application/json' \
    -d "{\"userId\":\"U-1001\",\"channel\":\"PUSH\",\"message\":\"알림 $i\"}"
  echo
done

동기 발행 응답 예시 — 같은 키는 같은 파티션

{"requestId":"7c1d...","partition":1,"offset":11}
{"requestId":"9a3f...","partition":1,"offset":12}
{"requestId":"b2e0...","partition":1,"offset":13}

검증 방법

1. 토픽이 선언대로 만들어졌는가

토픽 스펙 확인

cd kafka-lab
./kcli kafka-topics.sh --describe --topic notifications

기대 출력

Topic: notifications  TopicId: ...  PartitionCount: 3  ReplicationFactor: 3
	Configs: min.insync.replicas=2,retention.ms=259200000
	Topic: notifications	Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
	Topic: notifications	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
	Topic: notifications	Partition: 2	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2

2. 같은 키가 같은 파티션으로 가는가

키와 파티션을 함께 출력

./kcli kafka-console-consumer.sh --topic notifications --from-beginning \
  --timeout-ms 8000 \
  --property print.key=true --property print.partition=true \
  --property print.value=false 2>/dev/null | sort | uniq -c

같은 userId의 레코드가 모두 하나의 파티션 번호로 나옵니다. 기본 파티셔너는 키가 있으면 murmur2(key) % 파티션수로 결정적으로 배치합니다. 여기서 파티션이 여러 개로 흩어진다면 키가 null로 들어간 것입니다.

3. `concurrency`가 실제로 파티션을 나눠 갖는가

컨슈머 그룹 멤버 확인

./kcli kafka-consumer-groups.sh --describe --group notification-sender

기대 출력 — CONSUMER-ID 접미사(-0, -1, -2)가 스레드입니다

GROUP                TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID                                          CLIENT-ID
notification-sender  notifications  0          4               4               0    consumer-notification-sender-1-...                   consumer-notification-sender-1
notification-sender  notifications  1          13              13              0    consumer-notification-sender-2-...                   consumer-notification-sender-2
notification-sender  notifications  2          2               2               0    consumer-notification-sender-3-...                   consumer-notification-sender-3

concurrency: 3이므로 컨슈머 3개가 파티션 하나씩 가져갑니다. concurrency를 5로 올려 다시 확인해 보세요 — CONSUMER-ID는 5개가 되지만 파티션을 할당받은 것은 여전히 3개이고 2개는 아무 파티션도 없이 유휴 상태입니다.

4. 수동 ack가 유실을 막는가

acknowledge() 전에 프로세스를 죽여 봅니다

# NotificationListener.send() 안에 Thread.sleep(30000) 을 임시로 넣고 재기동한 뒤,
# 레코드를 하나 발행하고 처리 중에 프로세스를 SIGKILL 합니다.
curl -s -X POST http://localhost:8080/api/notifications/sync \
  -H 'Content-Type: application/json' \
  -d '{"userId":"U-9999","channel":"EMAIL","message":"ack 테스트"}'

pkill -9 -f 'com.example.notification.NotificationApplication'

# 오프셋이 전진하지 않았으므로 lag 이 1 로 남아 있습니다.
./kcli kafka-consumer-groups.sh --describe --group notification-sender | grep -v ' 0 '

# 재기동하면 같은 레코드를 다시 받아 처리합니다(at-least-once).
cd ../notification-service && mvn -q spring-boot:run

레코드가 사라지지 않고 다시 처리되는 것이 정답입니다. 만약 ack.acknowledge()를 처리 전으로 옮기면 같은 실험에서 그 레코드가 영구히 사라집니다(at-most-once). 커밋 시점이 유실·중복 경계를 결정한다는 것을 직접 확인할 수 있습니다 (예제 4에서 더 자세히 다룹니다).

프로덕션 고려사항

로컬 예제와 프로덕션의 차이
항목	이 예제	프로덕션
직렬화	`JsonSerializer` + record	Avro/Protobuf + Schema Registry. JSON은 필드 추가·삭제 시 컨슈머가 조용히 깨집니다(예제 7)
역직렬화 실패	처리하지 않음	`ErrorHandlingDeserializer`로 감싸지 않으면 잘못된 레코드 하나가 포이즌 필이 되어 파티션을 영구히 막습니다(예제 6)
에러 처리	없음 (예외가 그대로 나갑니다)	`DefaultErrorHandler` + DLQ를 반드시 구성합니다(예제 6)
발행 실패 대응	비동기는 로그만	유실이 곤란하면 outbox 패턴이나 동기 발행 + 재시도. 로그만 남기면 아무도 못 봅니다(예제 3)
멱등성	없음	at-least-once이므로 같은 알림이 두 번 발송될 수 있습니다. `requestId`를 게이트웨이의 멱등 키로 넘겨야 합니다
종료 처리	Spring 기본	Spring Boot는 컨테이너를 graceful하게 멈추지만, K8s `terminationGracePeriodSeconds`가 짧으면 SIGKILL로 커밋이 유실됩니다
모니터링	actuator 기본	Micrometer로 `kafka.consumer.`·`kafka.producer.`를 Prometheus에 노출하고 lag에 알림(예제 10)
보안	`PLAINTEXT`	`spring.kafka.security.protocol=SASL_SSL` + `spring.kafka.properties.sasl.*`. 자격증명은 시크릿에서 주입합니다

자주 하는 실수

케이스 1 — auto.offset.reset과 그룹 ID를 오해해 전체 재처리가 일어난 사고.
케이스 4 — 키를 지정하지 않아 순서가 뒤바뀐 사고.
케이스 5 — 파티션 수를 늘려 오히려 느려진 사고.

이어서 볼 곳

공식 문서 출처

Producer Configs — acks=all, batch.size=16384, buffer.memory=33554432, compression.type=none, linger.ms=5, enable.idempotence=true
Consumer Configs — enable.auto.commit=true, auto.offset.reset=latest, max.poll.records=500, max.poll.interval.ms=300000, session.timeout.ms=45000, heartbeat.interval.ms=3000, isolation.level=read_uncommitted
Topic Configs — min.insync.replicas=1, retention.ms=604800000
Spring Boot Reference — Apache Kafka Support — spring.kafka.* 프로퍼티, NewTopic 빈으로 토픽 생성
Spring for Apache Kafka Reference — Sending Messages — KafkaTemplate이 CompletableFuture를 반환한다는 서술
Spring for Apache Kafka Reference — Receiving Messages — @KafkaListener, AckMode 7종, Acknowledgment