kafka consumer API - Githubissues

Consumer API

public class SimpleConsumer {
    private final static Logger logger = LoggerFactory.getLogger(SimpleConsumer.class);
    private final static String TOPIC_NAME = "test";
    private final static String BOOTSTRAP_SERVERS = "my-kafka:9092";
    private final static String GROUP_ID = "test-group";

    public static void main(String[] args) {
        Properties configs = new Properties();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record: records) {
                logger.info("{}", record);
            }
        }
    }
}

컨슈머 그룹을 통해서 컨슈머의 목적을 구분 가능
무한 루프 내에서 poll 함수를 호출해서 카프카 클러스터로부터 데이터를 읽어온다.
poll 함수의 결과로 ConsumerRecord 리스트를 반호나한다. poll 메서드는 Duration 타입을 인자로 받는데, 이는 컨슈머 버퍼에 데이터를 기다리기 위한 타임아웃 간격을 의미한다.

컨슈머를 운영하는 2가지 방법

1개 이상의 컨슈머로 이루어진 컨슈머 그룹을 운영하는 것
토픽의 특정 파티션만 구독하는 컨슈머를 운영하는 것

컨슈머 그룹 운영

하나의 컨슈머는 토픽에 묶인 1개 이상의 파티션들에 할당되어 데이터를 가져갈 수 있다.
이 떄, 하나의 파티션은 컨슈머 그룹의 최대 1개의 컨슈머에게만 할당된다.
따라서 컨슈머 개수는 가져가고자 하는 토픽의 파티션 개수보다 같거나 작아야한다.
컨슈머 그룹은 다른 컨슈머 그룹과 격리되는특징이 있다.

컨슈머 그룹 예시 - 시스템 정보 확인 파이프라인

데이터를 카프카 브로커에 토픽, 3 파티션으로 저장
이를 별개의 컨슈머 그룹을 만들어서 엘라스틱 서치와 하둡에서 각각 처리
이렇게 하면 엘라스틱 서치나 하둡 둘 중 한군데에 장애가 나더라도 나머지 하나는 정상 동작함

컨슈머 리밸런싱

만일 컨슈머 그룹의 컨슈머가 장애를 일으킨다면 장애가 발생한 컨슈머에 할당된 파티션은 장애가 발생하지 않은 컨슈머에 소유권이 넘어간다. 이 과정을 리밸런싱이라 부른다.
컨슈머 추가 혹은 컨슈머 제외에서 이 리밸런싱이 일어남
카프카 브로커 중 한 대가 그룹 조정자 역할을 수행한다.
컨슈머는 브로커로부터 데이터를 어디까지 가져갔는지 커밋을 통해 기록한다.

오프셋 커밋

enable.auto.commit=true로 두면 poll() 메서드가 수행될 때 일정 간격마다 오프셋을 커밋한다.
그러나 컨슈머 강제 종료나 리밸런싱 등이 발생하면 데이터 중복이나 유실 발생이 가능하다.
데이터 중복이나 유실을 허용하지 않는 서비스라면 auto commit을 사용해서는 안된다.
명시적으로 오프셋 커밋을 수행하려면 poll() 메서드 호출 이후 반환 받은 데이터의 처리가 완료되고 commitSync() 메서드를 호출하면 된다.
비동기로 하고 싶으면 commitAsync() 함수를 호출할 것

컨슈머 내부구조

컨슈머는 poll() 호출 시점에 데이터를 가져오는 것이 아니라 내부적으로 Fetcher 인스턴스가 생성되어 미리 레코드들을 내부 큐로 가져온다.
poll() 메서드를 호출하면 컨슈머 내부 큐에 있는 레코드들을 반환받아 처리를 수행하게 된다.

컨슈머 필수 옵션

bootstrap.servers: 프카프카 클러스터에 속한 브로커의 호스트 이름:포트를 1개 이상 작성
key.serializer: 레코드의 메시지 키를 직렬화 하는 클래스를 지정
value.serializer: 레코드의 메시지 값을 직렬화하는 클래스 지정

컨슈머 선택 옵션

group.id: 컨슈머 그룹 아이디 지정, subscribe() 메서드로 토픽을 구독하여 사용할 때는 이 옵션이 필수
auto.offset.reset: 지정된 컨슈머 오프셋이 없는 경우 어느 오프셋부터 읽을지 선택하는 옵션, latest, earliest, none 중 1개로 설정
enable.auto.commit: 자동 커밋 or 수동 커밋
auto.commit.interval.ms: 오프셋 커밋 간격 지정
max.poll.records: poll() 메서드를 통해 반호나되는 레코드 개수 지정
session.timeout.ms: 컨슈머와 브로커와 연결이 끊기는 최대 시간
heartbeat.interval.ms: 하트비트 전송 시간 간격
max.poll.interval.ms: poll() 메서드를 호출하는 간격의 최대 시간 지정
isolation.level: 트랜잭션 프로듀서가 레코들르 트랜젝션 단위로 보낼 경우 사용

리밸런스 리스너를 가진 컨슈머

컨슈머가 추가 또는 제거되면 파티션을 컨슈머에 재할당하는 과정인 리밸런스가 일어난다.
poll() 메서드를 통해 반환받은 데이터를 모두 처리하기 전에 리밸런스가 발생하면 데이터를 중복 처리할 수 있다. 왜냐면 poll()로 받은 데이터를 처리하고 있는 단계고 아직 커밋하지 않았기 때문
떄문에 이러한 중복 처리를 방지하기 위해서는 리밸런스 발생 시 처리한 데이터를 기준으로 커밋을 시도해야 한다.
이를 위해 카프카 라이브러리는 ConsumerRebalanceListener 인터페이스를 지원한다. 이를 구현한 클래스는 onPartitionAssigned() 메서드와 onPartitionRevoked() 메서드이다.
onPartitionAssigned()는 리밸런스가 끝난 뒤 파티션이 할당 완료되면 호출되는 메서드
onPartitionRevoked()는 리밸런스가 시작되기 직전 호출 메서드

파티션을 직접 할당하는 컨슈머

컨슈머 운영시 subscribe() 메서드 외에 직접 파티션을 컨슈머에 명시적으로 할당도 가능

assign() 메서드를 이용하면 된다.

consumer.assign(Collections,singleton(new TopicPartition(TOPIC_NAME, PARTITION_NUMBER)));

컨슈머 안전 종료

KafkaConsumer 클래스의 wakeup() 메서드 사용

아래와 같이 wakeup() 메서드가 호출되면 WakeupException이 발생한다. 이를 예외처리한 뒤, finally 문에 consumer.close를 호출해주면 된다.

try {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record: records) {
                logger.info("{}", record);
                currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1, null));
                consumer.commitSync(currentOffsets);
            }
        }    
    } catch (WakeupException e) {
        logger.warn("Wakeup consumer");
    } finally {
        consumer.close();
    }

wakeup() 메서드 호출은 셧다운 훅을 사용하면 구현할 수 있다.


// shutdown hook 등록
Runtime.getRuntime().addShutdownHook(new ShutdownThread());

// wakeup 함수 호출하는 shutdown hook static class ShutdownThread extends Thread { public void run() { logger.info("Shutdown hook"); consumer.wakeup(); } }

yeomko22 / TIL

kafka consumer API #108