전달 보장과 트랜잭션

세 가지 전달 보장

공식 설계 문서는 세 가지를 다음과 같이 정의합니다.

공식 문서는 곧바로 중요한 단서를 붙입니다. "이것은 두 개의 문제로 갈라진다" — 메시지를 발행할 때의 내구성 보장과, 메시지를 소비할 때의 보장입니다. 그리고 exactly-once를 주장하는 많은 시스템에 대해 "작은 글씨를 읽어야 한다"고 경고합니다. 컨슈머나 프로듀서가 실패하는 경우, 컨슈머 프로세스가 여러 개인 경우, 디스크에 쓴 데이터가 유실될 수 있는 경우로 번역되지 않는 주장이 흔하다는 것입니다.

중복은 재시도에서, 유실은 커밋 순서에서

두 문제는 발생 지점이 다릅니다. 섞어서 생각하면 처방이 엉킵니다.

중복 — 프로듀서 쪽 재시도. 프로듀서가 메시지를 보낸 뒤 네트워크 오류를 만나면, 그 오류가 커밋 전에 발생한 것인지 커밋 후에 발생한 것인지 알 수 없습니다. 공식 문서는 이를 "자동 생성 키가 있는 데이터베이스 테이블에 insert하는 것과 비슷한 상황"으로 설명합니다. 재전송하면 중복이 생기고, 재전송하지 않으면 유실이 생깁니다. 0.11.0.0부터는 멱등 프로듀서가 이 딜레마를 해소합니다 — 브로커가 프로듀서마다 ID를 부여하고 시퀀스 번호로 중복을 제거합니다.

유실과 중복 — 컨슈머 쪽 커밋 순서. 공식 문서가 드는 두 가지 선택입니다.

1. 읽고 → 위치를 저장하고 → 처리한다. 위치 저장 후 처리 결과 저장 전에 크래시하면, 인수한 프로세스가 저장된 위치부터 시작하므로 그 앞의 몇 건은 처리되지 않은 채 건너뛰어집니다. → at-most-once
2. 읽고 → 처리하고 → 위치를 저장한다. 처리 후 위치 저장 전에 크래시하면, 인수한 프로세스가 이미 처리한 몇 건을 다시 받습니다. → at-least-once

공식 문서는 2번에 대해 "많은 경우 메시지에 기본 키가 있어 갱신이 멱등적이다" (같은 메시지를 두 번 받아도 같은 레코드를 덮어쓰는 것뿐)라고 덧붙입니다. 이것이 실무에서 at-least-once + 컨슈머 멱등성이 가장 널리 쓰이는 조합인 이유입니다.

보장별 설정 조합

멱등 프로듀서와 트랜잭션은 다른 것입니다

둘 다 "중복을 막는다"고 설명되지만 해결하는 문제가 다릅니다. 멱등성은 재시도로 인한 중복을 없애고, 트랜잭션은 여러 쓰기를 원자적으로 묶습니다.

멱등성이 요구하는 것

공식 문서는 enable.idempotence=true가 세 가지를 요구한다고 명시합니다.

• max.in.flight.requests.per.connection이 5 이하 — 이유가 명확합니다. 브로커가 프로듀서마다 최대 5개의 배치만 보관하기 때문입니다. 5를 넘으면 이전 배치가 브로커 쪽에서 제거되어 중복 판별을 할 수 없습니다.
• retries가 0보다 커야 합니다.
• acks가 all이어야 합니다.

충돌하는 설정이 있고 멱등성을 명시적으로 켜지 않았다면 멱등성이 비활성화됩니다. 명시적으로 켰는데 충돌 설정이 있으면 설정 오류가 됩니다 — 조용히 꺼지지 않습니다.

트랜잭션 — API와 코디네이터

공식 문서는 Kafka 트랜잭션이 다른 메시징 시스템의 트랜잭션과 다르다는 점을 먼저 짚습니다. Kafka에서는 컨슈머와 프로듀서가 분리되어 있고, 트랜잭션은 프로듀서만 갖습니다. 다만 프로듀서가 컨슈머의 위치(committed offset)를 트랜잭션 안에서 갱신할 수 있고, 바로 이것이 전체적인 exactly-once 동작을 만들어 냅니다.

transactional.id 의 역할

공식 문서의 정의입니다. transactional.id의 목적은 "단일 프로듀서 인스턴스의 여러 세션에 걸친 트랜잭션 복구를 가능하게 하는 것"입니다. 같은 transactional.id를 쓰는 트랜잭션이 새 트랜잭션 시작 전에 완료되었음을 보장하기 때문입니다. 재시작한 애플리케이션이 이전 인스턴스의 진행 중 트랜잭션을 중단(abort)시키는 효과도 여기서 나옵니다.

따라서 이 값은 파티션된 상태 저장 애플리케이션의 샤드 식별자에서 유도하는 것이 전형적이고, 애플리케이션 안에서 실행되는 프로듀서 인스턴스마다 유일해야 합니다.

API 호출 순서

공식 문서는 "모든 트랜잭션 API는 블로킹이며 실패 시 예외를 던진다"고 명시하고, send()에 콜백을 지정하거나 Future.get()을 호출할 필요가 없다고 합니다. 트랜잭션 중에 복구 불가한 오류가 발생하면 KafkaException이 던져지기 때문입니다.

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092");
// 이 값이 세션을 가로지르는 복구의 열쇠입니다. 샤드 식별자에서 결정적으로 유도하세요.
props.put("transactional.id", "order-enricher-shard-3");
Producer<String, String> producer =
        new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

// 한 번만 호출합니다. 같은 transactional.id 의 진행 중 트랜잭션을 정리합니다.
producer.initTransactions();

try {
    producer.beginTransaction();
    for (int i = 0; i < 100; i++) {
        producer.send(new ProducerRecord<>("orders-enriched",
                Integer.toString(i), Integer.toString(i)));
    }
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 복구 불가 — 프로듀서를 닫고 종료하는 것이 유일한 선택입니다.
    producer.close();
} catch (KafkaException e) {
    // 그 밖의 예외는 트랜잭션을 중단하고 다시 시도합니다.
    // abort 를 호출해야 이미 성공한 쓰기가 abort 로 표시됩니다.
    producer.abortTransaction();
}
producer.close();

4.x의 표준화된 예외 분류

Apache Kafka 4.x에서 트랜잭션 프로듀서의 오류 처리가 표준화되었습니다. 공식 문서가 정의하는 여섯 개 범주입니다. 어느 것이 애플리케이션까지 올라오는지가 핵심입니다.

트랜잭션 관련 설정

consume-transform-produce 패턴

Kafka 토픽을 읽어 처리하고 다른 Kafka 토픽에 쓰는 파이프라인이 EOS가 성립하는 정확한 범위입니다. 공식 문서는 그 원리를 이렇게 설명합니다 — 컨슈머의 위치가 내부 토픽의 메시지로 저장되므로, 출력 토픽에 쓰는 것과 같은 트랜잭션 안에서 오프셋을 Kafka에 쓸 수 있습니다. 트랜잭션이 중단되면 컨슈머의 저장된 위치는 이전 값으로 되돌아가고, 출력 토픽에 쓰인 데이터는 read_committed 컨슈머에게 보이지 않습니다.

공식 문서가 드는 세 가지 요건

프로듀서와 컨슈머를 직접 써서 exactly-once 처리를 만들 때 필요한 것입니다.

1. 컨슈머는 파티션 할당을 통해 자신이 그 파티션을 처리하는 그룹 내 유일한 컨슈머임을 보장받습니다.
2. 프로듀서는 트랜잭션을 사용해 자신이 생산하는 모든 레코드와 컨슈머를 대신해 갱신하는 오프셋을 원자적으로 처리합니다.
3. 리밸런싱과의 조합을 올바르게 다루기 위해, 컨슈머 인스턴스마다 프로듀서 인스턴스를 하나씩 쓰는 것이 권장됩니다. 더 복잡하고 효율적인 방식도 가능하지만 복잡도가 올라갑니다.

// ---- 컨슈머: 공식 문서가 요구하는 두 설정 ----
Properties cp = new Properties();
cp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
cp.put(ConsumerConfig.GROUP_ID_CONFIG, "order-enricher");
cp.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");   // 필수
cp.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");         // 필수
cp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
cp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// ---- 프로듀서: 컨슈머 인스턴스마다 하나 ----
Properties pp = new Properties();
pp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
pp.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-enricher-shard-3");
pp.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
pp.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(cp);
KafkaProducer<String, String> producer = new KafkaProducer<>(pp);

producer.initTransactions();
consumer.subscribe(List.of("orders"));

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    if (records.isEmpty()) continue;

    try {
        producer.beginTransaction();

        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        for (ConsumerRecord<String, String> record : records) {
            producer.send(new ProducerRecord<>("orders-enriched",
                    record.key(), enrich(record.value())));
            // 커밋할 오프셋은 "다음에 읽을 위치" = 현재 오프셋 + 1
            offsets.put(new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1));
        }

        // 오프셋 커밋을 트랜잭션에 포함시킵니다.
        // groupMetadata() 를 넘겨야 브로커가 제너레이션까지 검증해 좀비를 펜싱합니다.
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

        producer.commitTransaction();

    } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
        // 복구 불가 — 종료합니다.
        producer.close();
        consumer.close();
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();
        // ★ 중요: abort 후 컨슈머 위치를 커밋된 오프셋으로 되돌려야 재처리됩니다.
        //   자동으로 되감기지 않습니다.
        resetToLastCommitted(consumer, records.partitions());
    }
}

// 커밋된 오프셋을 다시 읽어 컨슈머 위치를 되돌립니다.
static void resetToLastCommitted(KafkaConsumer<?, ?> consumer,
                                 Set<TopicPartition> partitions) {
    Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(partitions);
    for (TopicPartition tp : partitions) {
        OffsetAndMetadata om = committed.get(tp);
        if (om != null) {
            consumer.seek(tp, om.offset());
        } else {
            consumer.seekToBeginning(List.of(tp));
        }
    }
}

Kafka Streams — processing.guarantee

공식 문서는 "Kafka에서 exactly-once 의미론을 얻는 가장 간단한 방법은 Kafka Streams를 쓰는 것"이라고 명시합니다. 위의 루프를 직접 쓰는 대신 설정 하나로 같은 결과를 얻습니다.

exactly_once_v2를 켜면 공식 문서 기준으로 다음이 자동으로 바뀝니다.

• commit.interval.ms 기본값이 100ms로 변경됩니다.
• 컨슈머에 isolation.level=read_committed가 설정됩니다.
• 프로듀서에 enable.idempotence=true가 설정됩니다.

그리고 기본적으로 최소 3대의 브로커가 필요합니다. 개발용으로는 transaction.state.log.replication.factor와 transaction.state.log.min.isr를 브로커 수에 맞춰 낮춰야 합니다. Streams 상세는 10장에서 다룹니다.

isolation.level 과 LSO

비트랜잭션 메시지는 두 모드 모두에서 무조건 반환됩니다. 그리고 메시지는 항상 오프셋 순서로 반환됩니다 — 이 두 사실이 다음 절의 근거입니다.

LSO — Last Stable Offset

공식 문서의 정의입니다. LSO는 "가장 처음 열려 있는 트랜잭션의 오프셋보다 1 작은 값"입니다. read_committed 모드의 poll()은 LSO까지만 반환합니다.

여기서 실무적으로 가장 중요한 결론이 나옵니다. 진행 중인 트랜잭션에 속한 메시지 뒤에 오는 메시지는, 그 자체가 이미 커밋되었더라도 해당 트랜잭션이 완료될 때까지 보류됩니다. 메시지가 오프셋 순서로만 반환되므로 중간을 뛰어넘을 수 없기 때문입니다. 그 결과 공식 문서 표현대로 "진행 중인 트랜잭션이 있으면 read_committed 컨슈머는 high watermark까지 읽을 수 없습니다."

컨트롤 레코드와 seekToEnd

트랜잭션이 커밋 또는 중단되면 트랜잭션 마커 레코드가 각 파티션에 기록됩니다. read_committed 컨슈머는 이 마커를 보고 중단된 트랜잭션의 레코드를 걸러 냅니다. 이 마커는 애플리케이션에 반환되지 않습니다 — 하지만 오프셋 공간은 차지합니다.

공식 문서가 명시하는 또 하나의 차이입니다. read_committed 모드에서 seekToEnd()는 high watermark가 아니라 LSO를 반환합니다. "끝으로 이동"의 의미가 모드에 따라 달라집니다.

EOS의 경계 — 이 장에서 가장 중요한 절

여기까지의 모든 장치는 Kafka 토픽을 읽어 Kafka 토픽에 쓰는 경계 안에서만 성립합니다. 공식 문서의 결론 문장을 그대로 인용하면 이렇습니다.

왜 외부 시스템에는 성립하지 않는가

공식 문서가 설명하는 이유는 명확합니다. 외부 시스템에 쓸 때의 제약은 "컨슈머의 위치와 실제로 저장된 출력을 조율해야 하는 필요"에 있습니다. 고전적 해법은 컨슈머 위치 저장과 출력 저장 사이의 2단계 커밋(two-phase commit)인데, 공식 문서는 "컨슈머가 쓰려는 많은 출력 시스템이 2단계 커밋을 지원하지 않는다"고 지적합니다.

결국 Kafka 트랜잭션은 Kafka 안의 파티션들과 __consumer_offsets에 대해서만 원자성을 만듭니다. HTTP 요청 하나, SQL INSERT 하나는 그 원자성 안에 들어갈 방법이 없습니다. 트랜잭션이 abort되어 Kafka 쪽이 전부 되돌아가도 이미 나간 HTTP 요청은 되돌아오지 않습니다.

그래서 컨슈머 멱등성을 직접 만듭니다

공식 문서가 제시하는 일반적 해법은 "컨슈머가 자신의 오프셋을 출력과 같은 곳에 저장하게 하는 것"입니다. 2단계 커밋보다 단순하고 범용적이기 때문입니다. 실무에서는 다음 세 가지 형태로 나타납니다.

// Kafka 에 오프셋을 커밋하지 않습니다. DB 가 진실의 원천입니다.
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { }
    @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition tp : partitions) {
            // DB 에 저장된 오프셋에서 재개합니다.
            Long offset = dao.loadOffset(tp.topic(), tp.partition());
            if (offset != null) consumer.seek(tp, offset);
            else consumer.seekToBeginning(List.of(tp));
        }
    }
});

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

    try (Connection conn = dataSource.getConnection()) {
        conn.setAutoCommit(false);
        for (ConsumerRecord<String, String> record : records) {
            dao.upsertOrder(conn, record.value());
            // 결과와 오프셋이 같은 트랜잭션에 들어갑니다 → 둘 다 저장되거나 둘 다 안 됩니다.
            dao.saveOffset(conn, record.topic(), record.partition(), record.offset() + 1);
        }
        conn.commit();
    } catch (SQLException e) {
        // 롤백되었으므로 다음 루프에서 같은 오프셋을 다시 읽습니다.
        log.error("DB 트랜잭션 실패 — 재처리됩니다", e);
    }
}

-- 유니크 제약이 중복 처리를 물리적으로 차단합니다.
CREATE TABLE processed_records (
    topic       VARCHAR(255) NOT NULL,
    partition_id INT         NOT NULL,
    record_offset BIGINT     NOT NULL,
    processed_at TIMESTAMP   NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (topic, partition_id, record_offset)
);

-- 부수효과보다 먼저 이력을 확정합니다.
-- INSERT 가 유니크 위반으로 실패하면 이미 처리한 것이므로 건너뜁니다.
INSERT INTO processed_records (topic, partition_id, record_offset)
VALUES ('payments', 3, 981274);

흔한 오해

시험 포인트 요약

확인 문제

EOS의 경계, isolation.level이 어느 쪽 설정인지, 멱등성과 트랜잭션의 구분이 집중적으로 나옵니다.

이어서 볼 곳

• 5장 · Consumer 심화 커밋 방식과 타임아웃 4종 — 이 장의 전제가 되는 내용입니다. CCDAK Application Development
• 4장 · Producer 심화 acks와 멱등성, 재시도 예산 — 프로듀서 쪽 상세. CCDAK Application Development
• 케이스 7 · EOS를 켰는데 중복 결제가 났다 EOS의 경계를 잘못 이해한 파이프라인의 실제 전개. 실수 케이스
• 예제 5 · consume-transform-produce EOS 동작하는 트랜잭션 파이프라인 전체 코드. 실무 예제
• 10장 · Kafka Streams exactly_once_v2로 같은 보장을 설정 하나로 얻는 방법. CCDAK Kafka Streams
• CCDAK 함정 사전 멱등성 vs 트랜잭션, read_uncommitted vs read_committed 비교표. 자격증

공식 문서 출처

이 장의 정의·설정 기본값·API 시그니처는 모두 아래에서 확인했습니다 (Apache Kafka 4.3 문서 기준).

• Design — Message Delivery Semantics — 세 보장의 정의, 중복/유실의 발생 지점, "기본은 at-least-once", 외부 시스템에 대한 EOS의 제약과 2단계 커밋 서술
• Design — Using Transactions — 프로듀서만 트랜잭션을 갖는 구조, EOS 3요건, 필수 설정(read_committed + enable.auto.commit=false + transactional.id), 4.x 표준화된 예외 6범주, abort 후 위치 리셋 요구
• Operations — Transaction Protocol — KIP-890 Transactions Server Side Defense, transaction.version 피처 플래그, 업그레이드/다운그레이드 동작
• Producer Configs — transactional.id, transaction.timeout.ms(60000), enable.idempotence(true)와 그 요구사항, max.in.flight.requests.per.connection(5)의 상한 이유, acks
• Consumer Configs — isolation.level(read_uncommitted), LSO 정의, seekToEnd의 동작 차이
• Broker Configs — transaction.state.log.replication.factor(3), transaction.state.log.min.isr(2), transaction.max.timeout.ms(900000), transactional.id.expiration.ms(604800000)
• Kafka Streams Configs — processing.guarantee의 값과 기본값, EOS 시 commit.interval.ms 100ms 변경, 자동 설정되는 컨슈머·프로듀서 설정, 브로커 3대 요구
• Javadoc — KafkaProducer — 트랜잭션 예시 코드, transactional.id의 목적과 유일성 요구, ProducerFencedException 처리, 멱등성의 세션 경계와 애플리케이션 재전송 경고
• Javadoc — Producer — sendOffsetsToTransaction(Map, ConsumerGroupMetadata) 시그니처