[Kafka] 컨슈머 오프셋 수동으로 커밋하기

개요

메시지 손실을 방지하기 위해 메시지 처리가 문제없이 완료되었을 경우에만 commit을 수행하도록, 컨슈머의 offset commit을 수동으로 설정하는 경우가 여러 존재합니다. 하지만 단순히 auto.offset.commit만 false로 지정하는 경우에는 원하는 방향으로 동작하지 않을 수도 있습니다.

이번 게시글에서는 자동 커밋의 동작 과정과 주의점, 그리고 커밋을 수동으로 제어하기 위한 auto.offset.commit 설정과 ack-mode 설정에 대해 알아보겠습니다.
 
 

auto.offset.commit

Kafka의 컨슈머는 읽은 메시지의 위치를 추적하기 위해 offset을 commit 하는 역할을 담당합니다. 이 offset은 Kafka의 내부 토픽인 __consumer_offsets에 저장되며, 컨슈머는 commit이 발생할 때마다 이 offset 값을 갱신합니다. 이후 컨슈머는 이 offset 값을 참조하여 다음에 처리할 레코드를 읽어옵니다.

컨슈머가 offset을 commit 하는 방식에는 자동 커밋과 수동 커밋이 있습니다. auto commit은 auto.offset.commit 설정이 true일 때 활성화되며, 기본적으로 이 값은 true로 설정되어 있습니다.

만약 자동 커밋이 활성화되어 있을 경우, poll() 메서드가 실행되었을 때 수행 시간이 auto.commit.interval.ms 설정값을 초과하는 경우 자동으로 offset이 commit 됩니다.

https://www.conduktor.io/kafka/delivery-semantics-for-kafka-consumers/

따라서 컨슈머는 poll을 요청할 때마다 auto.commit.interval.ms에 정의한 커밋 할 시간인지 아닌지 체크하게 되고, poll 요청으로 가져온 마지막 오프셋을 커밋합니다. 

실제로 Kafka의 소스 코드를 살펴보면, poll() 메서드 내부에서 maybeAutoCommitOffsetsAsync() 메서드가 호출되어 비동기적으로 auto commit이 일어나는 것을 확인할 수 있습니다. - ConsumerCoordinator.java #L1214

public void maybeAutoCommitOffsetsAsync(long now) {
    if (autoCommitEnabled) {
        nextAutoCommitTimer.update(now);
        if (nextAutoCommitTimer.isExpired()) {
            nextAutoCommitTimer.reset(autoCommitIntervalMs);
            doAutoCommitOffsetsAsync();
        }
    }
}

위 메서드의 동작 과정은 다음과 같습니다.

1. 자동 커밋 활성화 여부 확인: autoCommitEnabled 값을 통해 자동 커밋 기능이 활성화되어 있는지 확인합니다.
2. 타이머 업데이트 (nextAutoCommitTimer.update(now)): 현재 시각(now)을 기준으로 다음 자동 커밋이 예정된 타이머를 업데이트합니다. update 메서드는 현재 시간과 마지막 오프셋 커밋 시간 사이의 경과 시간을 계산하여, 다음 자동 커밋 실행까지 남은 시간을 조정합니다. 이 과정은 자동 커밋 주기를 관리하기 위한 것으로, 실제 커밋 작업이 시작되기 전에 현재 시간과 설정된 커밋 간격을 기반으로 타이머 상태를 최신화합니다.
3. 타이머 만료 여부 확인 (nextAutoCommitTimer.isExpired()): 업데이트된 타이머를 검사하여 설정된 자동 커밋 간격이 만료되었는지 확인합니다. 만약 타이머가 만료되었다면 (즉, auto.commit.interval.ms에 설정된 시간이 경과했다면), 다음 단계로 진행합니다.
4. 타이머 리셋 (nextAutoCommitTimer.reset(autoCommitIntervalMs)): 타이머를 리셋하여 다음 자동 커밋 시점을 현재 시각으로부터 설정된 간격(autoCommitIntervalMs) 이후로 조정합니다. 이는 자동 커밋이 실행된 후 다음 자동 커밋 주기를 시작하기 위한 준비 단계입니다.
5. 비동기 오프셋 커밋 실행 (doAutoCommitOffsetsAsync()): 실제로 오프셋 커밋을 비동기적으로 수행하는 메서드를 호출합니다. 이 단계에서 컨슈머가 최근에 처리한 메시지의 오프셋이 Kafka에 커밋됩니다.

 
 

자동 커밋 설정 시 주의점

자동 커밋 설정은 편리하지만 컨슈머 그룹의 리밸런싱이 발생할 때 메시지의 중복 처리나 유실 문제가 발생할 수 있습니다.

1. 메시지 유실

자동 커밋 설정이 1초로 지정된 경우를 상상해 보겠습니다. 컨슈머는 정상적으로 poll() 메서드를 통해 메시지를 가져오고 해당 메시지를 처리합니다.

그러나 메시지 처리 과정이 1초를 초과하는 시간이 걸릴 경우 문제가 발생합니다. 예를 들어 메시지를 DB에 저장하거나 다른 서비스에 전달하는 과정에서 네트워크 지연이나 장애 등 어떤 이유로든 처리 시간이 1초를 초과하였다고 가정해 봅시다. 이러한 상황에서 메시지 처리가 완료되기 전에 오프셋 커밋이 일어나고, 그 후 메시지 처리에서 에러가 발생하는 경우 해당 메시지는 다시 처리될 수 없게 됩니다. 즉 메시지 유실이 일어나게 됩니다.

2. 메시지 중복

다음과 같은 설정의 경우, 메시지의 중복 처리가 발생할 수 있습니다.

• enable.auto.commit = true
• auto.commit.interval.ms = 5000ms
• max.poll.records = 100

컨슈머는 poll() 메서드를 호출하여 최대 100개의 레코드를 가져옵니다. 그 후 이 중 30개의 레코드까지 처리한 상태에서 커밋이 이루어지기 전에 애플리케이션이 종료되거나 리밸런싱이 발생하면 컨슈머들이 재할당됩니다. 이후 컨슈머는 마지막으로 커밋된 offset부터 다시 데이터를 polling 하게 되는데, 이미 성공적으로 처리했던 30개의 레코드를 다시 중복 처리하게 됩니다.

메시지 유실의 경우 Dead Letter Queue(DLT)를 구성하여 처리하지 못한 메시지를 별도로 관리하고 재처리하는 방식으로 해결할 수 있고, 메시지 중복의 경우는 비즈니스 로직 내부에서 멱등성을 보장하는 로직을 통해 해결할 수 있습니다.

하지만 메시지 처리를 확인한 후에 수동으로 커밋 시점을 적절하게 조절한다면 사전에 이러한 문제를 조금이나마 방지할 수 있습니다.
 
 

수동 커밋 설정

Kafka 컨슈머의 커밋을 수동으로 설정하기 위해서는 auto.offset.commit 설정뿐만 아니라 ack-mode 설정까지 이루어져야 합니다.

먼저 Kafka 컨슈머 설정에서 ENABLE_AUTO_COMMIT_CONFIG를 false로 지정하여 자동 커밋 기능을 비활성화할 수 있습니다. enable.auto.commit 기본값은 true이기 때문에, false로 명시적으로 설정해야 자동 커밋 기능이 비활성화됩니다.

private Map<String, Object> consumerConfigs() {
    Map<String, Object> configs = new HashMap<>();
    // ...
    configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    return configs;
}

다음으로 @KafkaListener와 같은 리스너 컨테이너를 이용해 메시지를 소비한다고 가정하겠습니다.

@KafkaListener(topics = "test", groupId = "test")
public void listen(String message) {
    System.out.println("Received Message in group 'test': " + message);
}

위와 같이 설정하면 별도의 커밋을 수행하는 코드를 작성하지 않았으므로, 메시지는 소비되지만 offset commit은 발생하지 않을 것이라고 기대할 수 있습니다. 그러나 실제로는 enable.auto.commit을 false로 지정했음에도 불구하고 커밋이 정상적으로 이루어지며 Lag가 쌓이지 않습니다.

그 이유는 enable.auto.commit=false 설정에 의해 Kafka 클라이언트의 자동 커밋은 비활성 되지만, Spring은 기본적으로 오프셋 커밋을 활성화 상태로 두고 있기 때문입니다. 때문에 위 설정에서는 자동 커밋 때처럼 auto.commit.interval.ms의 영향은 받지는 않지만 리스너 메서드가 완료되면 자동으로 커밋이 일어나게 됩니다.

따라서 리스너 컨테이너의 Ackmode까지 별도로 수정해야만 진정한 의미의 수동 커밋이 이루어질 수 있습니다. AckMode는 설정을 위한 여러 방법이 있지만 아래처럼 ContainerFactory를 통해 간단하게 설정할 수 있습니다.

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> testKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> testFactory = new ConcurrentKafkaListenerContainerFactory<>();
    testFactory.getContainerProperties().setAckMode(AckMode.MANUAL);
    testFactory.setConsumerFactory(testConsumerFactory());
    return testFactory;
}

AcksMode	설명
RECORD	레코드 단위로 프로세싱 이후 커밋한다.
BATCH	poll() 메서드로 호출된 레코드가 모두 처리된 이후 커밋한다.
TIME	특정 시간 이후에 커밋한다. 이 옵션을 사용할 경우에는 시간 간격을 선언하는 `AckTime`옵션을 설정해야 한다.
COUNT	특정 개수만큼 레코드가 처리된 이후에 커밋한다. 이 옵션을 사용할 경우에는 레코드 개수를 선언하는 `AckCount`옵션을 설정해야 한다.
COUNT_TIME	TIME, COUNT 옵션 중 맞는 조건이 하나라도 나올 경우 커밋한다.
MANUAL	Acknowledgement.acknowledge() 메서드가 호출되면 다음번 poll()때 커밋한다. 매번 acknowledge() 메서드를 호출하면 BATCH 옵션과 동일하게 동작한다. 이 옵션을 사용할 경우 AcknowledgingMessageListener 또는 BatchAcknowledgingMessageListener를 리스너로 사용해야 한다.
MANUAL_IMMEDIATE	Acknowledgement.acknowledge() 메서드를 호출한 즉시 커밋한다. 이 옵션을 사용할 경우 AcknowledgingMessageListener 또는 BatchAcknowledgingMessageListener를 리스너로 사용해야 한다.

Ackmode의 default는 BATCH이기 때문에, enable.auto.commit=false로 지정하더라도 Listener 메서드가 종료될 때 스프링에서 자동으로 커밋이 수행하여 커밋이 일어나게 됩니다.

따라서 MANUAL 혹은 MANUAL_IMMEDIATE로 지정해야지만 원래 가정했던 것처럼 리스너에서 토픽을 소비하지만 offset commit은 일어나지 않고, Lag가 쌓이게 됩니다. 그리고 다음과 같이 Acknowledgement.acknowledge()를 호출하여 커밋 시점을 직접적으로 조절할 수 있습니다.

@KafkaListener(topics = "test", groupId = "test")
public void listen(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment) {
    System.out.println("Received Message in group 'test': " + record.value());
    // 메시지 처리 로직
    // ...
    // 처리가 완료된 후에 Acknowledgment를 사용하여 커밋
    acknowledgment.acknowledge();
}

 
 

참고

• https://stackoverflow.com/questions/61885717/kafka-consumer-with-enable-auto-commit-false-still-committing-the-offsets
• https://www.conduktor.io/kafka/delivery-semantics-for-kafka-consumers/