Kafka

커밋이란 무엇이고 왜 어긋나면 문제인가

poll()을 호출하면 그룹의 컨슈머들이 아직 읽지 않은, Kafka에 쓰인 레코드들이 반환된다. 즉 그룹의 컨슈머가 어떤 레코드까지 읽었는지 추적하는 수단이 있다는 뜻이다. Kafka의 독특한 특성 중 하나는 많은 JMS 큐처럼 컨슈머의 ack를 브로커가 추적하지 않는다는 점이다. 대신 컨슈머가 각 파티션에서 자신의 위치(오프셋)를 Kafka를 이용해 추적하게 한다. 파티션 내 현재 위치를 갱신하는 행위를 커밋이라 부른다.

컨슈머는 어떻게 오프셋을 커밋할까? 각 파티션의 커밋된 오프셋을 담아 특별한 __consumer_offsets 토픽에 메시지를 produce한다. 모든 컨슈머가 정상 동작 중이면 이것은 아무 영향이 없다. 그러나 컨슈머가 크래시하거나 새 컨슈머가 그룹에 들어오면 리밸런스가 일어나고, 리밸런스 후 각 컨슈머는 이전과 다른 파티션 집합을 할당받을 수 있다. 어디서부터 이어서 작업할지 알기 위해 컨슈머는 각 파티션의 마지막 커밋 오프셋을 읽고 거기서부터 계속한다.

여기서 두 가지 어긋남이 생길 수 있다.

커밋된 오프셋이 클라이언트가 실제로 처리한 마지막 메시지의 오프셋보다 작으면, 마지막 처리 오프셋과 커밋 오프셋 사이의 메시지들이 두 번 처리된다.

커밋된 오프셋이 실제로 처리한 마지막 메시지의 오프셋보다 크면, 그 사이의 모든 메시지가 컨슈머 그룹에서 누락된다.

그래서 오프셋 관리는 클라이언트 애플리케이션에 큰 영향을 준다.

자동 커밋 — 편하지만 통제가 없다

가장 쉬운 방법은 컨슈머에게 커밋을 맡기는 것이다. enable.auto.commit=true로 두면 5초마다 클라이언트가 poll()로 받은 가장 큰 오프셋을 커밋한다. 5초는 기본값이며 auto.commit.interval.ms로 조정한다.

중요한 것은 자동 커밋도 poll 루프가 구동한다는 점이다. poll할 때마다 컨슈머는 커밋할 시간이 되었는지 확인하고, 되었다면 직전 poll에서 반환한 오프셋을 커밋한다.

이 편의를 쓰기 전에 결과를 이해해야 한다. 기본적으로 자동 커밋은 5초마다 일어난다. 가장 최근 커밋 이후 3초가 지난 시점에 리밸런스가 발생했다고 하자. 리밸런싱 후 모든 컨슈머는 마지막으로 커밋된 오프셋부터 소비를 시작한다. 이 오프셋은 3초 전 것이므로, 그 3초 동안 도착한 모든 이벤트가 두 번 처리된다. 커밋 간격을 줄여 중복 구간을 좁힐 수는 있지만 중복을 완전히 없애는 것은 불가능하다.

자동 커밋이 켜져 있으면 poll 호출은 항상 이전 poll이 반환한 마지막 오프셋을 커밋한다. 어떤 이벤트가 실제로 처리되었는지는 알지 못한다. 그래서 poll()을 다시 호출하기 전에 poll()이 반환한 모든 이벤트를 반드시 처리해 두는 것이 결정적으로 중요하다(poll()과 마찬가지로 close()도 오프셋을 자동으로 커밋한다). 보통은 문제가 되지 않지만, 예외를 처리하거나 poll 루프를 중간에 빠져나갈 때 주의해야 한다.

정리하면 자동 커밋은 편리하지만 중복 메시지를 피할 만큼의 통제권을 개발자에게 주지 않는다.

commitSync — 가장 단순하고 신뢰할 수 있는 수동 커밋

대부분의 개발자는 커밋 시점을 더 통제하려 한다. 메시지 누락 가능성을 없애고, 리밸런싱 동안 중복되는 메시지 수를 줄이기 위해서다. 컨슈머 API는 타이머가 아니라 애플리케이션 개발자가 의미 있다고 판단한 시점에 현재 오프셋을 커밋하는 옵션을 제공한다.

enable.auto.commit을 false로 두면 애플리케이션이 명시적으로 선택할 때만 오프셋이 커밋된다. 커밋 API 중 가장 단순하고 신뢰할 수 있는 것이 commitSync()다. poll()이 반환한 마지막 오프셋을 커밋하고, 커밋이 완료되면 반환하며, 어떤 이유로든 커밋이 실패하면 예외를 던진다.

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        // 처리
    }
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        log.error("commit failed", e);
    }
}

기억할 점은 commitSync()가 poll()이 반환한 마지막 오프셋을 커밋한다는 것이다. 따라서 컬렉션의 모든 레코드 처리를 끝낸 뒤에 호출해야 한다. 그렇지 않으면 앞서 설명한 메시지 누락 위험이 생긴다. 리밸런스가 발생하면 가장 최근 배치의 시작부터 리밸런스 시점까지의 모든 메시지가 두 번 처리된다.

commitSync는 복구 불가능한 오류가 아닌 한 커밋을 계속 재시도한다. 재시도로도 안 되면 오류를 로깅하는 것 외에 할 수 있는 일이 별로 없다.

commitAsync — 빠르지만 재시도하지 않는다

수동 커밋의 단점은 브로커가 커밋 요청에 응답할 때까지 애플리케이션이 블로킹된다는 것이다. 이는 애플리케이션 처리량을 제한한다. 커밋 빈도를 줄여 처리량을 개선할 수는 있지만, 그러면 리밸런스가 만들어 낼 잠재적 중복이 늘어난다.

다른 선택지가 비동기 커밋 API다. 브로커의 응답을 기다리지 않고 요청만 보내고 계속 진행한다.

consumer.commitAsync();

단점은 commitSync()가 성공하거나 재시도 불가능한 실패를 만날 때까지 재시도하는 것과 달리, commitAsync()는 재시도하지 않는다는 점이다.

왜 재시도하지 않을까? commitAsync()가 서버 응답을 받을 시점에는 이미 더 나중의 커밋이 성공했을 수 있기 때문이다. 오프셋 2000을 커밋하는 요청을 보냈는데 일시적 통신 장애로 브로커가 요청을 받지 못해 응답도 없었다고 하자. 그 사이 우리는 다음 배치를 처리하고 오프셋 3000을 성공적으로 커밋했다. 이때 commitAsync()가 실패한 커밋을 재시도하면, 이미 3000이 커밋된 뒤에 2000을 커밋하는 데 성공할 수 있다. 리밸런스가 일어나면 이는 더 많은 중복을 만든다.

commitAsync()는 브로커가 응답할 때 호출될 콜백을 넘길 수 있다. 커밋 오류를 로깅하거나 메트릭으로 세는 데 흔히 쓰인다. 콜백에서 재시도를 하려면 위의 커밋 순서 문제를 반드시 의식해야 한다.

비동기 재시도의 순서를 맞추는 단순한 패턴이 있다. 단조 증가하는 시퀀스 번호를 두고, 커밋할 때마다 번호를 올려 그 시점의 번호를 commitAsync 콜백에 함께 넘긴다. 재시도를 보내려 할 때 콜백이 받은 시퀀스 번호가 인스턴스 변수와 같으면 더 새로운 커밋이 없었다는 뜻이므로 재시도해도 안전하다. 인스턴스 쪽 번호가 더 크면 이미 더 새로운 커밋이 나갔으므로 재시도하지 않는다.

동기와 비동기 커밋 결합하기

일시적인 문제라면 재시도 없이 커밋이 한 번 실패해도 큰 문제가 아니다. 다음 커밋이 성공할 것이기 때문이다. 하지만 이것이 컨슈머를 닫기 직전 또는 리밸런스 직전의 마지막 커밋이라면 반드시 성공시켜야 한다.

그래서 흔히 쓰는 패턴이 평상시에는 commitAsync()를 쓰고 종료 직전에 commitSync()를 호출하는 것이다.

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            // 처리
        }
        consumer.commitAsync();
    }
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

정상 동작 중에는 commitAsync를 쓴다. 더 빠르고, 한 커밋이 실패해도 다음 커밋이 재시도 역할을 한다. 그러나 종료 시점에는 "다음 커밋"이 없으므로, 성공하거나 복구 불가능한 실패를 만날 때까지 재시도하는 commitSync()를 호출한다.

특정 오프셋 커밋 — 배치 중간에 커밋하기

마지막 오프셋만 커밋하면 배치 처리를 끝낼 때마다만 커밋할 수 있다. poll()이 거대한 배치를 반환했는데 리밸런스 시 전부 재처리하지 않으려면 배치 중간에 커밋하고 싶어진다. 이때 commitSync()나 commitAsync()를 그냥 호출하면 안 된다. 아직 처리하지 않은 마지막 반환 오프셋을 커밋해 버리기 때문이다.

다행히 컨슈머 API는 커밋하고 싶은 파티션과 오프셋의 맵을 인자로 받는 commitSync()/commitAsync()를 제공한다. 토픽 "customers"의 파티션 3에서 마지막으로 받은 메시지 오프셋이 5000이라면 그 파티션에 대해 5000을 커밋할 수 있다. 컨슈머가 여러 파티션을 소비할 수 있으므로 모든 파티션의 오프셋을 직접 추적해야 하고, 그만큼 코드가 복잡해진다.

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0;

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        // 처리
        currentOffsets.put(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1, "no metadata"));
        if (count % 1000 == 0)
            consumer.commitAsync(currentOffsets, null);
        count++;
    }
}

각 레코드를 읽은 뒤 오프셋 맵을 갱신하는데, 값이 record.offset() + 1이라는 점이 핵심이다. 이는 "다음에 처리할 것으로 기대하는 메시지의 오프셋", 즉 다음에 읽기 시작할 위치다. 예제는 1,000 레코드마다 커밋하지만 시간 기준이나 레코드 내용 기준으로 커밋해도 된다. 여기서는 commitAsync를 썼지만 commitSync도 완전히 유효하며, 특정 오프셋을 커밋할 때도 앞서 본 오류 처리는 그대로 필요하다.

ConsumerRebalanceListener — 파티션을 잃기 직전에 할 일

컨슈머는 종료 직전, 그리고 파티션 리밸런싱 직전에 정리 작업을 하고 싶어 한다. 파티션 소유권을 곧 잃는다는 것을 안다면 마지막으로 처리한 이벤트의 오프셋을 커밋해야 한다. 이벤트를 버퍼에 모아 가끔씩만 처리하는 구조라면 소유권을 잃기 전에 쌓인 이벤트를 처리해야 하고, 파일 핸들이나 DB 커넥션을 닫아야 할 수도 있다.

컨슈머 API는 파티션이 추가되거나 제거될 때 실행할 코드를 등록할 수 있게 해 준다. subscribe()를 호출할 때 ConsumerRebalanceListener를 함께 넘기면 된다. 이 인터페이스에는 두 메서드가 있다.

onPartitionsRevoked(Collection<TopicPartition> partitions)는 리밸런싱이 시작되기 전, 그리고 컨슈머가 메시지 소비를 멈춘 후에 호출된다. 여기가 오프셋을 커밋할 자리다. 이 파티션을 다음에 받을 쪽이 어디서 시작할지 알 수 있게 하기 위해서다.

onPartitionsAssigned(Collection<TopicPartition> partitions)는 파티션이 재할당된 후, 컨슈머가 메시지 소비를 시작하기 전에 호출된다.

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    }
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync(currentOffsets);
    }
}

consumer.subscribe(topics, new HandleRebalance());

세부 사항 두 가지가 중요하다. 첫째, 여기서 커밋하는 것은 "우리가 처리한 최신 오프셋"이지 "아직 처리 중인 배치의 마지막 오프셋"이 아니다. 배치 중간에 파티션이 회수될 수 있기 때문이다. 둘째, 곧 잃을 파티션만이 아니라 모든 파티션의 오프셋을 커밋한다. 이미 처리된 이벤트에 대한 오프셋이므로 그렇게 해도 해가 없다. 그리고 리밸런스가 진행되기 전에 커밋이 확실히 끝나도록 commitSync()를 쓴다.

그리고 가장 중요한 것은, 리스너를 subscribe() 메서드에 넘겨야 실제로 호출된다는 점이다.

seek()와 외부 저장소에 오프셋 두기

지금까지는 각 파티션의 마지막 커밋 오프셋부터 소비했지만, 다른 오프셋에서 읽기 시작하고 싶을 때가 있다. 파티션 처음부터 읽으려면 seekToBeginning(TopicPartition tp), 끝으로 건너뛰어 새 메시지만 읽으려면 seekToEnd(TopicPartition tp)가 있다. API는 특정 오프셋으로 seek하는 것도 지원한다. 몇 개 뒤로 돌아가거나, 뒤처진 시간 민감 애플리케이션이 더 관련 있는 메시지로 건너뛰는 데 쓸 수 있다.

가장 흥미로운 유스케이스는 오프셋을 Kafka가 아닌 다른 시스템에 저장하는 경우다.

흔한 시나리오를 보자. 애플리케이션이 Kafka에서 이벤트(예: 웹사이트 클릭스트림)를 읽어 처리하고, 결과를 DB나 NoSQL, Hadoop에 저장한다. 데이터를 잃고 싶지도 않고 같은 결과를 DB에 두 번 저장하고 싶지도 않다.

레코드마다 처리 후 오프셋을 커밋해도, 레코드가 DB에 저장된 뒤 오프셋 커밋 전에 애플리케이션이 크래시하면 그 레코드는 다시 처리되고 DB에 중복이 생긴다. 이는 레코드와 오프셋을 하나의 원자적 동작으로 저장할 수 있다면 피할 수 있는 문제다. 그러나 레코드는 DB에, 오프셋은 Kafka에 쓰는 한 그것은 불가능하다.

해법은 레코드와 오프셋을 둘 다 DB에, 하나의 트랜잭션으로 쓰는 것이다. 그러면 레코드 처리가 끝나 오프셋이 커밋되었거나, 아니면 레코드가 재처리되거나 둘 중 하나가 된다.

남는 문제는 오프셋이 Kafka가 아니라 DB에 있을 때 컨슈머가 파티션을 할당받으면 어디서부터 읽을지 어떻게 아느냐는 것이다. 바로 여기에 seek()를 쓴다. 컨슈머가 시작할 때 또는 새 파티션을 할당받을 때, DB에서 오프셋을 조회해 그 위치로 seek()하면 된다.

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        commitDBTransaction();
    }
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions)
            consumer.seek(partition, getOffsetFromDB(partition));
    }
}

consumer.subscribe(topics, new SaveOffsetsOnRebalance(consumer));
consumer.poll(0);
for (TopicPartition partition : consumer.assignment())
    consumer.seek(partition, getOffsetFromDB(partition));

동작 순서가 중요하다. 컨슈머가 처음 시작하면 토픽 구독 후 poll()을 한 번 호출해 컨슈머 그룹에 조인하고 파티션을 할당받은 다음, 곧바로 할당된 파티션들에 대해 seek()로 올바른 오프셋으로 이동한다. seek()는 소비 위치만 갱신하므로 다음 poll()이 올바른 메시지를 가져온다. seek()에 오류가 있으면(예: 존재하지 않는 오프셋) 그 예외는 poll()에서 던져진다.

오프셋과 데이터를 외부 저장소에 두고 exactly-once 시맨틱을 구현하는 방법은 여러 가지지만, 모두 ConsumerRebalanceListener와 seek()를 사용해 오프셋이 제때 저장되고 컨슈머가 올바른 위치에서 읽기 시작하도록 보장해야 한다.

poll 루프를 안전하게 빠져나가기 — wakeup()

poll 루프를 종료하기로 결정하면 다른 스레드에서 consumer.wakeup()을 호출해야 한다. 컨슈머 루프를 메인 스레드에서 돌린다면 ShutdownHook에서 호출하면 된다.

핵심은 consumer.wakeup()이 다른 스레드에서 호출해도 안전한 유일한 컨슈머 메서드라는 점이다. wakeup을 호출하면 poll()이 WakeupException을 던지며 빠져나온다. 스레드가 poll에서 대기 중이 아닐 때 wakeup()이 호출되었다면, 다음 반복에서 poll()을 호출할 때 예외가 던져진다.

WakeupException 자체는 처리할 필요가 없다(애플리케이션이 예기치 않게 종료되지 않도록 잡아 주기만 하면 된다). 다만 스레드를 종료하기 전에 반드시 consumer.close()를 호출해야 한다. 컨슈머를 닫으면 필요한 경우 오프셋을 커밋하고, 그룹 코디네이터에게 그룹을 떠난다는 메시지를 보낸다. 그러면 코디네이터가 즉시 리밸런싱을 유발하므로, 세션 타임아웃을 기다린 뒤에야 이 컨슈머의 파티션이 다른 컨슈머에게 배정되는 상황을 피할 수 있다.

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        consumer.wakeup();
        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

ShutdownHook은 별도 스레드에서 실행되므로 우리가 취할 수 있는 유일하게 안전한 동작이 wakeup을 호출해 poll 루프를 깨는 것이다.

역직렬화기 — 프로듀서와 짝을 맞춰야 한다

프로듀서가 객체를 바이트 배열로 바꾸는 직렬화기를 필요로 하듯, 컨슈머는 Kafka에서 받은 바이트 배열을 Java 객체로 바꾸는 역직렬화기를 필요로 한다.

당연하게 들리지만 결정적으로 중요한 사실은, 이벤트를 produce할 때 쓴 직렬화기와 consume할 때 쓸 역직렬화기가 일치해야 한다는 것이다. IntSerializer로 직렬화하고 StringDeserializer로 역직렬화하면 결과가 좋을 수 없다. 즉 개발자가 각 토픽에 어떤 직렬화기가 사용되었는지 추적해야 하고, 각 토픽에는 사용할 역직렬화기가 해석할 수 있는 데이터만 들어가도록 관리해야 한다.

이것이 Avro와 스키마 저장소를 쓰는 이점 중 하나다. AvroSerializer는 특정 토픽에 쓰이는 모든 데이터가 그 토픽의 스키마와 호환됨을 보장할 수 있고, 따라서 대응하는 역직렬화기와 스키마로 역직렬화될 수 있음을 보장한다. 프로듀서 쪽이든 컨슈머 쪽이든 호환성 오류는 적절한 오류 메시지로 쉽게 잡히므로, 직렬화 오류를 찾겠다고 바이트 배열을 디버깅할 필요가 없다.

커스텀 역직렬화기는 커스텀 직렬화기의 로직을 뒤집는 것이다(ByteBuffer로 감싸 int를 읽고 이름 바이트를 읽어 객체를 재구성). 하지만 이 방식은 프로듀서와 컨슈머를 강하게 결합시키고 취약하며 오류를 유발하기 쉬우므로 권장되지 않는다. JSON, Thrift, Protobuf, Avro 같은 표준 메시지 포맷이 더 나은 해법이다. 게다가 컨슈머 쪽에도 Customer 클래스 구현이 필요하고, 클래스와 직렬화기가 produce 쪽과 consume 쪽에서 일치해야 하므로, 데이터를 공유하는 컨슈머와 프로듀서가 많은 큰 조직에서는 이것만으로도 난제가 된다.

Avro 역직렬화는 KafkaAvroDeserializer를 value.deserializer로 지정하고 schema.registry.url을 함께 설정하면 된다. 이렇게 하면 컨슈머는 프로듀서가 등록한 스키마를 사용해 메시지를 역직렬화한다.

스탠드얼론 컨슈머 — 그룹 없이 쓰기

지금까지는 파티션이 자동으로 할당되고 컨슈머 추가·제거 시 자동으로 리밸런스되는 컨슈머 그룹을 다뤘다. 보통은 이것이 원하는 동작이지만, 훨씬 단순한 것이 필요할 때가 있다. 단일 컨슈머가 항상 토픽의 모든 파티션 또는 특정 파티션에서 데이터를 읽어야 한다는 것을 이미 알고 있는 경우다. 이럴 때는 그룹도 리밸런스도 필요 없다. 컨슈머에게 특정 토픽·파티션을 할당하고, 메시지를 소비하고, 가끔 오프셋을 커밋하면 된다.

어떤 파티션을 읽어야 할지 정확히 안다면 토픽을 subscribe하지 않고 자신에게 파티션을 assign한다. 중요한 제약은 컨슈머가 토픽을 구독해 컨슈머 그룹의 일원이 되거나, 자신에게 파티션을 할당하거나 둘 중 하나만 가능하고 동시에는 안 된다는 것이다.

List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic");
if (partitionInfos != null) {
    for (PartitionInfo partition : partitionInfos)
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
    consumer.assign(partitions);
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        // 처리
        consumer.commitSync();
    }
}

먼저 클러스터에 그 토픽의 파티션 목록을 물어본다(특정 파티션만 소비할 계획이면 이 단계는 건너뛸 수 있다). 원하는 파티션을 정했으면 그 목록으로 assign()을 호출한다.

리밸런스가 없고 파티션을 직접 찾아야 한다는 점을 제외하면 나머지는 평소와 같다. 다만 기억할 것이 하나 있다. 누군가 토픽에 새 파티션을 추가해도 이 컨슈머는 통보받지 못한다. 주기적으로 consumer.partitionsFor()를 확인하거나, 파티션이 추가될 때 애플리케이션을 재시작하는 방식으로 직접 처리해야 한다.