Producer 심화

전송 파이프라인 — send()에서 콜백까지

KafkaProducer.send()는 브로커에 데이터를 보내지 않습니다. 레코드를 직렬화하고 파티션을 정한 뒤 메모리 버퍼에 넣고 즉시 반환합니다. 실제 네트워크 전송은 별도의 Sender 스레드가 합니다. 이 분리가 Producer 동작 전체의 출발점입니다.

단계별로 무슨 일이 일어나는가

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public final class OrderProducer {

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

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");   // 초기 연결 이중화
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 아래 세 개는 Kafka 4.x 기본값이므로 생략해도 동일합니다.
        // 의도를 코드에 남기기 위해 명시했습니다.
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        // try-with-resources 로 닫아야 버퍼에 남은 배치가 flush 됩니다.
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {

            ProducerRecord<String, String> record =
                new ProducerRecord<>("orders", "order-1024", "{\"amount\":12900}");

            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // 콜백은 Sender 스레드에서 실행됩니다. 여기서 블로킹하면 전송 전체가 멈춥니다.
                    // 로깅과 메트릭 증가까지만 하고, 재처리는 큐로 넘깁니다.
                    log.error("send failed topic={} key={}", record.topic(), record.key(), exception);
                    return;
                }
                log.debug("sent partition={} offset={}", metadata.partition(), metadata.offset());
            });

            // 종료 직전 명시적 flush — close() 도 내부적으로 flush 합니다.
            // try-with-resources 를 쓰지 않는다면 shutdown hook 에서
            // producer.close(Duration.ofSeconds(30)) 을 호출해야 버퍼가 비워집니다.
            producer.flush();
        } catch (RuntimeException e) {
            log.error("producer terminated abnormally", e);
            throw e;
        }
    }
}

콜백은 언제, 어느 스레드에서 실행되는가

Callback 인터페이스의 공식 javadoc은 두 문장으로 핵심을 정리합니다 — 이 콜백은 일반적으로 백그라운드 I/O 스레드에서 실행되므로 빨라야 하고, 레코드가 서버에서 acknowledge된 시점에 호출됩니다.

콜백의 계약

• 실행 스레드는 Sender(I/O) 스레드입니다. 애플리케이션 스레드가 아닙니다. 콜백 안에서 DB 호출·HTTP 호출·락 획득을 하면 그 프로듀서의 모든 파티션 전송이 함께 지연됩니다.
• (metadata, exception) 중 한쪽은 항상 null입니다. 예외가 있을 때 metadata는 모든 필드가 -1인 빈 메타데이터가 전달됩니다(null이 아닙니다). 파티션을 정할 수 없었던 경우에도 -1이 들어갑니다.
• 같은 파티션으로 보낸 레코드들의 콜백은 순서가 보장됩니다. 공식 javadoc의 예시대로 callback1이 callback2보다 먼저 실행됩니다. 다른 파티션 사이에는 순서 보장이 없습니다.
• 호출 시점은 브로커 응답을 받았을 때 또는 최종 실패가 확정되었을 때입니다. 최종 실패의 상한이 delivery.timeout.ms입니다.

배치와 지연 — batch.size × linger.ms

RecordAccumulator는 파티션별로 배치를 만듭니다. 하나의 배치가 전송되는 조건은 두 가지이고 먼저 충족되는 쪽이 이깁니다.

1. 그 파티션에 batch.size(기본 16384바이트) 분량이 쌓임
2. linger.ms(기본 5ms) 경과

공식 문서의 표현은 명확합니다 — "한 파티션에 대해 batch.size 분량의 레코드를 얻으면 이 설정과 무관하게 즉시 전송하고, 그보다 적으면 지정한 시간만큼 linger하며 더 기다린다."

batch.size의 미묘한 성질 세 가지

• batch.size보다 큰 레코드는 배치되지 않습니다. 공식 문서가 "이 크기보다 큰 레코드를 배치하려는 시도는 하지 않는다"고 명시합니다. 큰 레코드는 자기 혼자 하나의 요청이 됩니다.
• batch.size=0은 배치를 완전히 끕니다. 처리량이 크게 떨어지므로 진단 목적 외에는 쓰지 않습니다.
• 버퍼는 지정한 크기만큼 미리 할당됩니다. batch.size를 과하게 크게 두면 메모리를 낭비합니다.

buffer.memory와 역압

buffer.memory(기본 33554432 = 32MiB)는 전송 대기 레코드를 담는 전체 버퍼 크기입니다. 생산 속도가 전송 속도보다 빠르면 이 버퍼가 차고, 그때 send()가 max.block.ms만큼 블로킹된 뒤 TimeoutException으로 실패합니다. 공식 문서는 압축과 in-flight 요청 유지에도 추가 메모리가 쓰이므로 이 값이 프로듀서 전체 메모리의 엄격한 상한은 아니라고 명시합니다.

acks와 내구성 — min.insync.replicas와의 결합

세 값의 정확한 의미

min.insync.replicas와의 조합

min.insync.replicas는 토픽·브로커 설정이고 acks는 프로듀서 설정입니다. 소속이 다릅니다. 공식 문서는 이 설정이 acks=all일 때만 효력을 가진다고 명시합니다. 미달 시 프로듀서는 NotEnoughReplicasException 또는 NotEnoughReplicasAfterAppendException을 받습니다 — 둘 다 retriable입니다.

멱등성 프로듀서 — 무엇을 보장하고 무엇을 보장하지 않는가

재시도는 중복을 만듭니다. 프로듀서가 배치를 보냈고 브로커가 기록까지 했는데 응답이 유실되면, 프로듀서는 실패로 판단해 같은 배치를 다시 보냅니다. 결과는 로그에 두 번 기록된 같은 레코드입니다. 멱등성 프로듀서는 이것을 막습니다.

PID와 시퀀스 번호

브로커는 각 프로듀서에게 producer id(PID)를 부여하고, 프로듀서는 파티션마다 단조 증가하는 시퀀스 번호를 붙여 배치를 보냅니다. 2장에서 본 레코드 배치 헤더의 producerId · producerEpoch · baseSequence가 바로 이것입니다.

브로커는 들어온 배치의 첫·마지막 시퀀스 번호를 그 프로듀서의 마지막 값과 비교해 중복을 판정합니다. 이미 기록한 시퀀스면 기록하지 않고 성공으로 응답합니다. 프로듀서는 재시도가 성공한 것으로 보고 진행하고, 로그에는 사본이 하나만 남습니다.

멱등성이 요구하는 세 조건

공식 문서는 멱등성 활성화의 전제 조건을 명시합니다.

멱등성의 경계

순서 보장 — in-flight와 재시도의 관계

순서가 깨지는 메커니즘은 단순합니다. 같은 파티션으로 배치 A와 B를 연달아 보냈는데 A가 실패해 재시도되고 B가 먼저 성공하면, 로그에는 B가 앞에 기록됩니다. 공식 문서는 retries 설명에서 이 시나리오를 그대로 서술합니다.

retries는 이제 조절 대상이 아닙니다

retries의 기본값은 2147483647(Integer.MAX_VALUE)입니다. 공식 문서는 "이 설정은 그대로 두고 대신 delivery.timeout.ms로 재시도 동작을 제어하는 것을 일반적으로 권장한다"고 명시합니다. 즉 실질적인 상한은 횟수가 아니라 시간입니다.

파티셔너 — 키가 있을 때와 없을 때

키가 있으면 — 결정적 해시

공식 문서는 기본 파티셔닝 로직을 이렇게 설명합니다 — "파티션이 지정되지 않았고 키가 있으면 키의 해시를 기준으로 파티션을 고른다." 실제 구현은 Kafka 소스의 BuiltInPartitioner에 있고, 계산식은 다음과 같습니다.

// 직렬화된 키 바이트 → murmur2 해시 → 양수화 → 파티션 수로 나눈 나머지
partition = Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions;

세 가지 결론이 여기서 나옵니다.

• 같은 키는 항상 같은 파티션으로 갑니다. 그래서 키 단위 순서가 보장됩니다.
• 대상은 직렬화된 키 바이트입니다. 직렬화기를 바꾸면(예: String → Avro) 같은 논리적 키가 다른 파티션으로 갈 수 있습니다.
• 파티션 수(numPartitions)가 나눗셈에 들어갑니다. 파티션을 24개에서 32개로 늘리면 대부분의 키가 다른 파티션으로 이동하고, 그 순간 키 단위 순서 보장이 끊깁니다 (케이스 5).

키가 없으면 — sticky 배치, 라운드로빈이 아닙니다

공식 문서의 서술입니다 — "파티션도 키도 없으면 그 파티션에 batch.size 이상이 생산될 때 바뀌는 sticky 파티션을 고른다." 같은 문서는 기본 전략을 "한 파티션에 최소 batch.size 바이트가 생산될 때까지 그 파티션으로 레코드를 보내는 전략"으로 요약합니다.

즉 레코드마다 파티션을 바꾸는 것이 아니라, 배치가 찰 때까지 한 파티션에 몰아넣고 그다음 다른 파티션으로 전환합니다. 목적은 배치 효율입니다 — 레코드를 파티션마다 흩뿌리면 배치가 작게 조각나 요청 수가 늘고 압축률도 떨어집니다.

파티셔너 관련 설정

커스텀 파티셔너

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;

import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * VIP 고객의 레코드를 마지막 파티션에 격리하고, 나머지는 기본과 동일하게
 * murmur2 해시로 분배합니다. 격리 목적은 일반 트래픽 폭주가 VIP 처리 지연에
 * 영향을 주지 않게 하는 것입니다.
 */
public class VipIsolatingPartitioner implements Partitioner {

    private Set<String> vipKeys = Set.of();

    @Override
    public void configure(Map<String, ?> configs) {
        Object raw = configs.get("vip.customer.ids");
        if (raw != null) {
            vipKeys = Set.of(raw.toString().split(","));
        }
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (numPartitions == 0) {
            throw new IllegalStateException("no partition metadata for topic " + topic);
        }

        int vipPartition = numPartitions - 1;

        if (keyBytes == null) {
            // 키가 없으면 VIP 판별이 불가능합니다. VIP 파티션을 제외한 범위로 보냅니다.
            // (기본 sticky 배치의 이점은 포기하게 됩니다 — 그래서 키를 항상 넣는 편이 낫습니다)
            return numPartitions == 1 ? 0
                 : Utils.toPositive(Utils.murmur2(new byte[0])) % vipPartition;
        }

        if (vipKeys.contains(key.toString())) {
            return vipPartition;
        }

        // VIP 파티션을 제외한 나머지에 기본과 같은 방식으로 분배
        return numPartitions == 1 ? 0
             : Utils.toPositive(Utils.murmur2(keyBytes)) % vipPartition;
    }

    @Override
    public void close() {
        // 정리할 자원이 없습니다.
    }
}

압축 — 배치 단위로 적용됩니다

compression.type의 프로듀서 기본값은 none이고 허용 값은 none, gzip, snappy, lz4, zstd입니다.

핵심은 공식 문서의 이 문장입니다 — "압축은 데이터의 전체 배치 단위로 수행되므로, 배치가 잘 되는지가 압축 효율에도 영향을 준다." 즉 batch.size와 linger.ms를 올리면 압축률도 함께 올라갑니다. 압축만 켜고 배치 설정을 그대로 두면 기대한 효과가 나오지 않습니다.

토픽 레벨 compression.type

토픽에도 같은 이름의 설정이 있고 기본값은 producer입니다. 허용 값은 uncompressed, zstd, lz4, snappy, gzip, producer입니다. producer는 프로듀서가 지정한 코덱을 그대로 유지한다는 뜻입니다.

에러 처리 — retriable, non-retriable, 그리고 시간 예산

시간 예산의 포함 관계

delivery.timeout.ms(기본 120000ms)는 공식 문서의 정의대로 send()가 반환한 뒤 성공 또는 실패를 보고하기까지의 상한입니다. 여기에 세 가지가 모두 포함됩니다.

1. 전송 전에 배치를 기다린 시간 (linger.ms)
2. 브로커 응답을 기다린 시간 (request.timeout.ms, 기본 30000ms)
3. 재시도 가능한 실패로 인한 재시도 전체 시간 (retry.backoff.ms 기본 100, retry.backoff.max.ms 기본 1000)

그래서 공식 문서는 delivery.timeout.ms가 request.timeout.ms + linger.ms의 합보다 크거나 같아야 한다고 명시합니다.

# 요청 하나가 브로커 응답을 기다리는 시간
request.timeout.ms=30000

# 배치 대기
linger.ms=5

# 전체 예산: 30s 요청 × 재시도 + 배치 대기 <= 180s
# 30s 요청이 대략 5~6회 반복될 수 있는 예산입니다.
delivery.timeout.ms=180000

# 재시도 간 백오프 (지수 증가, 상한 1s)
retry.backoff.ms=100
retry.backoff.max.ms=1000

# retries 는 건드리지 않습니다 — 기본값(사실상 무한)을 두고 시간으로 제어합니다

retriable과 non-retriable

판정 기준은 예외가 RetriableException을 상속하는지입니다. Apache Kafka 4.3 소스의 계층 구조는 다음과 같습니다.

KafkaException
├── SerializationException                      ← non-retriable
└── ApiException
    ├── RetriableException                      ← 프로듀서가 자동 재시도
    │   ├── TimeoutException
    │   ├── NotEnoughReplicasException
    │   ├── NotEnoughReplicasAfterAppendException
    │   ├── CorruptRecordException
    │   └── RefreshRetriableException            ← 메타데이터 갱신 후 재시도
    │       └── InvalidMetadataException
    │           ├── LeaderNotAvailableException
    │           ├── NotLeaderOrFollowerException
    │           ├── UnknownTopicOrPartitionException
    │           └── NetworkException
    ├── ApplicationRecoverableException          ← 애플리케이션이 복구해야 함
    │   └── ProducerFencedException
    ├── RecordTooLargeException                  ← non-retriable
    ├── OutOfOrderSequenceException              ← non-retriable
    └── InvalidConfigurationException
        ├── InvalidTopicException                ← non-retriable
        ├── RecordBatchTooLargeException         ← non-retriable
        └── AuthorizationException
            └── TopicAuthorizationException      ← non-retriable

Callback 인터페이스의 공식 javadoc은 콜백에 전달될 수 있는 예외를 두 그룹으로 나열합니다. 시험 대비로는 이 목록이 가장 정확한 근거입니다.

void publish(Order order) {
    ProducerRecord<String, Order> record =
        new ProducerRecord<>("orders", order.customerId(), order);
    try {
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                sentCounter.increment();
                return;
            }
            // 비동기 실패: 재시도 예산을 모두 소진했거나 non-retriable 입니다.
            // 여기서 다시 send() 하지 마세요 — Sender 스레드를 재귀적으로 막습니다.
            if (exception instanceof org.apache.kafka.common.errors.RetriableException) {
                // delivery.timeout.ms 를 다 쓰고도 실패 → 클러스터 쪽 문제로 취급
                deadLetterQueue.offer(order);
            } else {
                // 데이터 자체가 문제 → 다시 보내도 같은 결과입니다
                poisonPillStore.save(order, exception);
            }
        });
    } catch (org.apache.kafka.common.errors.SerializationException e) {
        // 직렬화 실패는 호출 스레드에서 동기적으로 발생합니다.
        poisonPillStore.save(order, e);
    } catch (org.apache.kafka.common.errors.TimeoutException e) {
        // 메타데이터 미확보 또는 buffer.memory 소진 (max.block.ms 초과)
        backpressureSignal.raise();
        throw e;
    }
}

전송 파이프라인 관측 — 어느 메트릭이 어디를 보는가

앞의 파이프라인 각 단계는 서로 다른 메트릭으로 관측됩니다. 어떤 메트릭이 어느 구간을 덮는지 알아야 “느리다”는 증상을 단계로 좁힐 수 있습니다. CCDAK의 Application Observability 도메인이 이 구분을 묻습니다.

Producer 설정 정리

기본값은 Apache Kafka 4.3 공식 문서(Producer Configs) 기준입니다. 전체 목록과 프리셋은 설정 치트시트에 있습니다.

목적별 프리셋

처리량 우선 프리셋의 구체적 값은 워크로드에 따라 달라집니다. 위 범위는 방향을 나타내는 출발점이며, 공식 문서가 제시하는 권장값은 아닙니다. 측정 절차는 예제 3에 있습니다.

흔한 오해

시험 포인트 정리

확인 문제

CCDAK Application Development 도메인의 중심 내용입니다. 단일 선택 · 복수 선택 · 연결형 · 순서 배열이 섞여 있으며 키보드만으로 완주할 수 있습니다.

이어서 볼 곳

• 5장 · Consumer 심화 poll 루프, 타임아웃 4종, 리밸런스, KIP-848 새 프로토콜. CCDAK 최우선
• 6장 · 전달 보장과 트랜잭션 멱등성과 트랜잭션의 차이, EOS의 경계, read_committed와 LSO. CCDAK 최우선
• 예제 3 · 무손실 프로듀서 돌아가는 코드로 acks·minISR·재시도 예산을 직접 검증합니다. 실무 예제
• 케이스 4 · 키 없이 보낸 이벤트 순서가 뒤바뀐 상태 변경 이벤트. 원인 추적과 복구. 케이스 스터디
• 케이스 3 · 프로듀서 전송 실패의 진짜 원인 ISR 부족과 unclean leader election이 프로듀서에 어떻게 보이는가. 케이스 스터디
• 설정 치트시트 Producer 설정 전체 표와 처리량·지연·내구성 3가지 프리셋. 빠른참조

공식 문서 출처

이 장의 설정 기본값·동작 서술·예외 목록은 모두 아래에서 확인했습니다 (Apache Kafka 4.3 기준).

• Producer Configs — 이 장의 모든 프로듀서 설정 기본값과 설명
• Producer Configs — acks — 세 값의 정확한 정의, acks=0에서 오프셋 -1과 retries 무효, all = -1
• Producer Configs — linger.ms — 기본값이 4.0에서 0 → 5로 변경된 사실과 근거
• Producer Configs — batch.size — 크기 초과 레코드는 배치하지 않음, 0이면 배치 비활성, 브로커 backpressure 영향
• Producer Configs — enable.idempotence — in-flight 5 이하 · retries > 0 · acks=all 요구, 충돌 시 ConfigException 또는 조용한 비활성화
• Producer Configs — max.in.flight.requests.per.connection — 브로커가 프로듀서별로 최대 5개 배치를 보관하는 이유
• Producer Configs — delivery.timeout.ms — request.timeout.ms + linger.ms 이상 요구
• Producer Configs — partitioner.class — 기본 sticky 전략 서술, RoundRobinPartitioner, Partitioner 인터페이스
• Producer Configs — max.block.ms — 메타데이터·버퍼 대기 포함, 사용자 직렬화기·파티셔너 내부 블로킹 미포함
• Topic Configs — compression.type — 기본값 producer의 의미와 허용 값
• Topic Configs — min.insync.replicas — acks=all에서 ISR 전원 ack 요구, NotEnoughReplicas / NotEnoughReplicasAfterAppend
• Design — Message Delivery Semantics — 멱등 전달, PID와 시퀀스 번호로 중복 제거
• Design — Availability and Durability Guarantees — acks=all이 "현재 ISR 전원" 기준이라는 서술과 예시
• Implementation — Record Batch — producerId · producerEpoch · baseSequence, 중복 판별에 첫·마지막 시퀀스를 쓰는 방식
• Javadoc — Callback — 백그라운드 I/O 스레드 실행, 예외 시 metadata의 -1, retriable / non-retriable 예외 목록
• Javadoc — KafkaProducer.send() — 같은 파티션 콜백의 순서 보장
• Upgrading Apache Kafka — 4.0에서 DefaultPartitioner · UniformStickyPartitioner · NotLeaderForPartitionException 제거, Partitioner.onNewBatch 제거