Kafka producer: acks, idempotence와 batching

1. 개요

Kafka에 메시지를 보내는 것은 단순히 send() 한 줄이면 된다. 어려운 부분은 따라오는 질문들에 있다: 메시지가 정말 저장되었는가? Producer가 재시도할 때 중복으로 기록되지 않는가? 그리고 초당 수만 개의 메시지를 보내면서 broker를 막지 않으려면 어떻게 해야 하는가?

이 글에서는 이 세 질문에 답하는 세 가지 구성 그룹을 다룬다: acks, enable.idempotence, 그리고 batching (batch.size, linger.ms, compression.type). 이 글을 읽은 후에는 순수 Kafka client와 Spring Boot 3.x 모두에서 안전하고 빠른 producer를 구성하는 방법을 알게 되고, 이를 테스트하는 방법도 배운다.

2. 설정

순수 Java의 경우, kafka-clients만 필요하다:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

Spring Boot 3.x의 경우 spring-kafka를 사용한다. Spring Boot에서 관리하는 버전이며, kafka-clients를 포함한다:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

이 글 전체에서 우리는 주문 이벤트를 보낸다:

public record Order(String id, String customerId, BigDecimal total) {

    public String toJson() {
        return """
            {"id":"%s","customerId":"%s","total":%s}""".formatted(id, customerId, total);
    }
}

3. acks: 메시지가 언제 전송된 것으로 간주되는가?

acks는 producer가 요청을 성공으로 간주하기 전에 broker로부터 몇 개의 확인을 기다려야 하는지를 결정한다.

값	의미	위험
acks=0	아무것도 기다리지 않음	메시지 손실을 모르고 지남
acks=1	Leader가 쓰기를 완료할 때까지 기다림	Leader가 죽기 전에 follower가 복사하지 못하면 메시지 손실
acks=all (-1)	ISR의 모든 replica가 쓰기를 완료할 때까지 기다림	약간 더 높은 latency

순수 client로 구성하는 예:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");

try (var producer = new KafkaProducer<String, String>(props)) {
    var order = new Order("o-1001", "c-42", new BigDecimal("250000"));
    RecordMetadata meta = producer
        .send(new ProducerRecord<>("orders", order.customerId(), order.toJson()))
        .get();
    System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
}

.get()을 호출하면 전송이 동기화된다. 여기서 이렇게 하는 것은 단지 결과를 출력하기 위함이다. acks=all일 때, get()은 ISR의 모든 replica가 쓰기를 완료한 후에만 반환된다:

partition=2 offset=17

우리는 customerId를 key로 사용하므로 같은 고객의 모든 주문이 같은 partition으로 가고 정확한 순서를 유지한다.

3.1. acks=all이 min.insync.replicas가 없으면 충분하지 않음

acks=all은 “ISR에 현재 있는 모든 replica”를 의미한다. ISR이 leader만 남으로 축소되었다고 가정하자. 이 경우 acks=all은 acks=1과 다르지 않다. Topic 또는 broker 레벨에서 추가 구성이 필요하다:

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2

replication-factor=3이고 min.insync.replicas=2일 때, 각 쓰기마다 최소 2개의 복사본이 필요하다. ISR이 2 아래로 떨어지면, producer는 단일 복사본으로 몰래 쓰는 대신 NotEnoughReplicasException 오류를 받는다.

참고: min.insync.replicas를 replication-factor와 같게 설정하지 말자. 이렇게 하면 broker를 재시작하는 것만으로도 topic이 모든 쓰기를 거부한다.

4. Idempotence: 재시도해도 중복으로 기록되지 않음

acks=all은 재시도를 수반한다. 다음 시나리오를 생각해보자: Broker가 메시지를 이미 썼지만 응답이 도중에 손실되었다. Producer는 전송이 실패했다고 생각하고 다시 보내므로 메시지가 두 번 쓰인다.

Idempotent producer는 이 시나리오를 정확히 해결한다. enable.idempotence=true를 활성화하면:

• Broker는 producer에게 **producer ID (PID)**를 할당한다.
• Partition으로 전송되는 각 batch는 증가하는 sequence number를 포함한다.
• Broker는 이미 본 sequence를 가진 batch를 버리고, sequence를 건너뜀 batch를 거부한다. 따라서 많은 요청이 진행 중일 때도 순서가 유지된다.

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

각 줄의 의미:

• ENABLE_IDEMPOTENCE_CONFIG는 PID 및 sequence number 메커니즘을 활성화한다.
• Idempotence는 acks=all을 의무화한다. enable.idempotence=true와 함께 acks=1을 명시적으로 설정하면, producer는 초기화 시 ConfigException을 던진다.
• max.in.flight.requests.per.connection은 ≤ 5여야 한다. 이 한계 내에서 Kafka는 여전히 partition 내 순서를 보장한다.
• retries를 매우 크게 설정한다. 실제 재시도 시간은 delivery.timeout.ms에 의해 제한된다 (기본값 2분).

Kafka 3.0부터 acks=all과 enable.idempotence=true가 기본값이다. 그럼에도 불구하고 명시적으로 선언해야 한다. 그렇지 않으면 누군가 acks=1을 설정하는 것만으로도 idempotence가 조용히 비활성화된다.

4.1. Idempotence가 할 수 없는 것

여기가 자주 오해받는 부분이다:

• 그것은 단지 한 producer 세션 내에서, 각 partition에 대해 중복을 방지한다. 애플리케이션이 재시작되면 producer는 새로운 PID를 받는다. 재시작 후 다시 보낸 메시지는 여전히 중복으로 기록된다.
• 그것은 당신의 코드가 send()를 두 번 호출하는 것으로 인한 중복을 방지하지 않는다. 예를 들어 catch에서 자체 재시도할 때.
• 그것은 여러 partition에 원자적 쓰기를 제공하지 않는다.

위의 경우들은 transaction (transactional.id) 또는 orderId에 따라 중복을 제거할 수 있는 consumer가 필요하다.

5. Batching: Latency와 throughput의 트레이드오프

Producer는 각각의 메시지를 하나씩 보내지 않는다. send()는 단지 record를 메모리 내 buffer에 배치하고, 각 partition별로 정리한다. 백그라운드 thread (sender)는 여러 record를 batch로 모아서 보낸다. 세 가지 주요 매개변수:

매개변수	기본값	역할
batch.size	16384 byte	Partition에 대한 batch의 최대 크기
linger.ms	0 (Kafka 4.0부터 5)	Batch를 채우기 위해 기다릴 시간
compression.type	none	전체 batch 압축: gzip, snappy, lz4, zstd
buffer.memory	32 MB	대기 중인 모든 batch에 대한 총 메모리

Batch는 batch.size를 채우거나 linger.ms가 만료될 때 전송된다. 둘 중 먼저 오는 조건에 따라. linger.ms=0일 때, batch는 거의 항상 즉시 전송되므로 일반적으로 몇 개의 record만 포함한다.

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

이 구성은 최대 약 10ms의 추가 latency를 허용한다. 대신, 각 요청은 압축된 64KB의 데이터를 포함한다. 주문 이벤트처럼 반복이 많은 JSON 데이터의 경우, lz4 또는 zstd는 보통 네트워크 대역폭과 broker의 디스크 용량을 크게 줄인다. Batch 압축은 각 메시지를 압축하는 것보다 훨씬 효율적이다. Batch가 클수록 활용할 수 있는 반복 문자열이 많아지기 때문이다.

Producer의 메트릭을 통해 검증할 수 있다:

producer.metrics().forEach((name, metric) -> {
    if (name.name().equals("batch-size-avg") || name.name().equals("records-per-request-avg")) {
        System.out.println(name.name() + " = " + metric.metricValue());
    }
});

linger.ms=0으로 10,000개의 주문을 연속으로 보내고, linger.ms=10으로 반복하면 records-per-request-avg가 뚜렷이 증가하는 것을 볼 수 있다:

# linger.ms=0
batch-size-avg = 1830.4
records-per-request-avg = 21.7
# linger.ms=10
batch-size-avg = 41022.9
records-per-request-avg = 486.2

정확한 숫자는 시스템과 부하에 따라 다르지만, 추세는 항상 그렇다.

참고: Buffer가 가득 차면 (consumer가 느린 것은 관계없고, 여기서는 broker가 느리거나 네트워크가 막힌 경우), send()은 최대 max.block.ms (기본값 60초) 동안 block되었다가 TimeoutException을 던진다. HTTP 요청 처리 thread에서 send()을 호출할 때 이 가능성을 계산하자.

참고: delivery.timeout.ms는 linger.ms + request.timeout.ms 이상이어야 한다. 그렇지 않으면 producer는 초기화를 거부한다.

6. Spring Boot 3.x에서의 구성

Spring Boot는 일반적인 매개변수에 대한 고유한 속성이 있다. 나머지 매개변수는 properties를 통한다:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      batch-size: 64KB
      compression-type: lz4
      buffer-memory: 64MB
      properties:
        enable.idempotence: true
        linger.ms: 10
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000

Publisher는 KafkaTemplate을 사용한다. Spring Kafka 3.0부터 send()는 CompletableFuture를 반환한다:

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, String>> publish(Order order) {
        return kafkaTemplate.send("orders", order.customerId(), order.toJson())
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Order {} 전송 실패", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Order {} -> partition {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

세 가지 점에 주의하자:

• Callback은 producer의 sender thread에서 실행된다. whenComplete 내에서 무거운 작업을 하거나 blocking을 호출하지 말자. 다른 모든 전송을 느리게 할 것이다.
• Callback이 오류를 받으면, producer는 delivery.timeout.ms 한계 내에서 재시도를 완료했다. 손으로 다시 send()을 호출하는 식으로 추가 재시도를 하지 말자. 이것이 바로 idempotence가 막을 수 없는 중복의 원인이다.
• CompletableFuture를 반환하면 호출자가 기다릴지 여부를 스스로 결정할 수 있다.

7. Producer 행동 테스트

kafka-clients는 broker 없이 테스트하기 위한 MockProducer를 제공한다. 이를 사용하려면 로직을 Producer 인터페이스를 받는 클래스로 분리해야 한다:

public class OrderSender {

    private final Producer<String, String> producer;

    public OrderSender(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> send(Order order) {
        return producer.send(new ProducerRecord<>("orders", order.customerId(), order.toJson()));
    }

    public static Properties reliableConfig(String bootstrap) {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        return p;
    }
}

JUnit 5와 AssertJ를 사용한 테스트:

class OrderSenderTest {

    @Test
    void whenSendingOrders_thenKeyIsCustomerId() {
        var mock = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        sender.send(new Order("o-1", "c-42", new BigDecimal("100")));
        sender.send(new Order("o-2", "c-42", new BigDecimal("200")));

        assertThat(mock.history())
            .hasSize(2)
            .extracting(ProducerRecord::key)
            .containsOnly("c-42");
    }

    @Test
    void whenBrokerFails_thenFutureCompletesExceptionally() {
        var mock = new MockProducer<>(false, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        Future<RecordMetadata> future = sender.send(new Order("o-3", "c-7", BigDecimal.TEN));
        mock.errorNext(new NotEnoughReplicasException("ISR < min.insync.replicas"));

        assertThatThrownBy(future::get)
            .hasCauseInstanceOf(NotEnoughReplicasException.class);
    }

    @Test
    void reliableConfig_isAccepted() {
        var props = OrderSender.reliableConfig("localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        var config = new ProducerConfig(props);

        assertThat(config.getString(ProducerConfig.ACKS_CONFIG)).isEqualTo("-1");
        assertThat(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isTrue();
    }
}

각 테스트는 하나의 행동을 확인한다:

• 첫 번째 테스트는 autoComplete=true를 사용하므로 모든 send()가 즉시 성공한다. history()를 사용하면 key를 확인할 수 있고, 따라서 고객별 순서가 유지되는지 확인할 수 있다.
• 두 번째 테스트는 autoComplete=false를 사용한 후 errorNext(...)로 broker가 쓰기를 거부하도록 시뮬레이션한다. 이것이 코드의 오류 처리를 테스트하는 방법이다.
• 마지막 테스트는 ProducerConfig를 초기화하여 Kafka가 구성 조합을 자체 검증하도록 한다. 잘못된 조합, 예를 들어 acks=1과 idempotence를 함께 사용하면 배포 시 발생하는 대신 ConfigException을 던져서 테스트가 실패한다. Kafka는 all을 "-1"로 정규화하므로 assertion은 "-1"과 비교한다.

8. 언제 느슨하게 할 것인가?

“최대 안전” 구성이 항상 맞는 것은 아니다:

• 메트릭, 로그, clickstream: 몇 기록의 손실은 허용된다. 이 경우 acks=1, 높은 linger.ms, 그리고 최고의 throughput을 위한 zstd 압축을 사용한다.
• 메시지별 낮은 latency가 필요한 시스템 (예: 실시간 가격 정보): linger.ms를 작게 유지한다 (0–2ms). 이 경우 큰 batch를 기대하지 말자.
• 결제, 주문: 항상 acks=all, idempotence, 그리고 min.insync.replicas=2를 사용한다. 정확히 한 번의 읽기-처리-쓰기가 필요하면, transaction을 고려해야 한다.

참고: max.in.flight.requests.per.connection=1을 설정하는 것은 idempotence가 이미 활성화되어 있을 때 “보험”이 아니다. 그것은 단지 throughput을 감소시킨다.

9. 결론

세 가지 구성 그룹은 세 가지 다른 문제를 해결한다:

• acks=all과 min.insync.replicas의 조합은 메시지 손실을 방지한다.
• enable.idempotence는 재시도가 중복을 일으키지 않도록 하고 partition 내의 순서를 유지한다. 그 범위는 단일 producer 세션이다.
• batch.size, linger.ms, compression.type은 throughput을 결정하고, 약간의 latency를 대가로 한다.

이러한 값들을 명시적으로 선언하자. 기본값에 의존하지 말자. Test로 검증하여 배포 시 놀라움을 피하자. 그리고 producer가 이미 당신을 위해 하고 있으므로 코드에서 send()을 손으로 재시도하지 말자.