TIL

6.5 정확히 한 번 컨슈머 동작

• 카프카 프로듀서의 ‘정확히 한 번 전송’을 간단히 복습해보자
  • 브로커 측에는 전체 트랜잭션을 관리하는 트랜잭션 코디네이터가 존재
  • 코디네이터는 정확히 한 번 전송이 성공하면 해당 레코드의 트랜잭션 성공을 표시하는 특별한 메시지를 추가
  • 컨슈머는 이 특수한 메시지만 읽는다면 정확히 한 번 읽을 수 있다.
• 아래는 컨슈머의 정확히 한 번 읽기 위한 설정이다.

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class ExactlyOnceConsumer {

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String groupId = "exactly-once-consumer-group";
        String topic = "input-topic";

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 오프셋 자동 커밋 끄기
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); // 트랜잭션 커밋 메시지만 읽기
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topic));

        // 실제 poll/consume 처리
        while (true) {
            // 실제 메시지 읽기 로직 추가
        }
    }
}

• ISOLATION_LEVEL_CONFIG를 read_commited로 설정함으로써 트랜잭션 컨슈머로 동작시킬 수 있다.
  • 기본값은 read_uncommitted (모든 메시지를 읽을 수 있음)
• 트랜잭션 컨슈머라고 해서 정확히 한 번만 가져오는 것은 아니다.
  • 컨슈머의 경우 프로듀서와 달리 트랜잭션 코디네이터와 통신하지 않는다.
  • 트랜잭션 프로듀서가 보낸 메시지만 가져올 수 있는지에 대해서만 옵션으로 선택 가능
  • 즉 컨슘된 메시지가 다른 싱크 저장소로 중복 저장될 수 있다.
    • 컨슈머가 한 번만 읽더라도 다른 애플리케이션에 메시지를 저장하는 과정에서 중복 처리될 수도 있다.
• 컨슈머 동작까지 정확히 한 번 처리가 가능하려면 ‘컨슘-메시지 처리-프로듀싱’ 동작이 모두 한 트랜잭션이어야 한다.
  • 트랜잭션 처리 과정에서 실패가 발생하면, 해당 컨슈머 그룹의 커밋 오프셋을 증가시키지 않게 해야 한다.
  • 일부 컨슈머 애플리케이션에선 이 기능을 지원하는 경우도 있으므로 확인이 필요하다.
    • ex) 카프카 커넥터 중 HDFS 커넥터의 경우 정확히 한 번 HDFS에 저장하도록 지원한다.