April 14, 2022 5 min to read

kafka 프로듀서 정리

프로듀서에 대해서

kafka 프로듀서 간단 정리

프로듀서

프로듀서는 카프카 브로커로 데이터를 전송할 때 내부적으로 파티셔너, 배치 생성 단계를 거친다.

프로듀서 전송 과정

ProducerRecord –(send())–> Partitioner –> Accumulator(batches in Topics) –> Sender –> Kafka Cluster

ProducerRecord

메시지 전송을 위해 ProducerRecord 객체를 이용한다.

• 파티션 번호 직접 지정
• 타임스탬프 설정
• 메시지 키 설정

ProducerRecord에서는 메시지 키가 무엇인지, 커스텀 파티셔너인지 여부를 따지기 전에 가장 먼저 파티션 번호가 있는지를 우선 순위로 확인하고 처리한다.

프로듀서에서 데이터를 전송할 파티션을 선정하는 방법은 3가지가 있다.

1. 파티션 번호를 직접 지정하여 지정된 파티션에 바로 데이터를 전송
2. 가장 우선 순위인 파티션 번호가 지정되어 있지 않는 경우에 메시지 키를 설정하여 메시지 키의 해시 값을 이용해 데이터를 전송할 파티션을 지정한다.
3. 파티션 번호, 메시지 키 모두 지정돼있지 않는 경우에는 라운드 로빈 방식으로 데이터를 전송할 파티션을 할당한다.

메시지 키가 설정된 메시지는 무조건 동일한 파티션에 지정된다(하나의 파티션을 독점하는 것이 아닌 여러 메시지와 같은 파티션을 사용할 수는 있지만 같은 메시지 키를 가진 메세지는 무조건 같은 파티션에 지정)

Partitioner 종류

• UniformStickyPartitioner
• RoundRobinPartitioner

파티션 번호 지정 혹은 메시지 키가 설정 돼있을 때는 지정된 파티션 혹은 메시지 키의 해시값과 파티션을 매칭하여 데이터를 전송한다는 공통점이 있다.

프로듀서 주요 옵션

• acks : 프로듀서가 전송한 데이터가 브로커들에 정상적으로 저장되었는지 전송 성공 여부를 확인하는 용도의 옵션

• buffer.memory(default: 32MB) : broker로 전송할 데이터를 배치로 모으기 위해 설정한 버퍼 메모리 용량 설정
• retries(default: 2147483647) : 프로듀서가 브로커로부터 에러를 받고 난 뒤 재전송을 시도하는 횟수를 지정
• batch.size(default: 16384) : 배치로 전송할 레코드 최대 용량 지정
• linger.ms(default: 0) : 배치를 전송하기 전까지 대기하는 최소 시간

배치가 모두 묶일 때 까지 기다렸다가 한번에 모두 동일한 파티션으로 데이터를 전송하는 방식의 UniformStickyPartitioner의 linger.ms도 0인건가?

• partitioner.class(default: DefaultParitioner) : 레코드를 파티션에 전송할 때 적용하는 파티셔너 클래스를 지정
• enable.idempotence(default: false)

프로듀서와 브로커 사이의 네트워크 장애로 한 번 이상 데이터가 전송되더라도 이를 PID(Produder unique ID)외에 시퀀스 넘버가 추가되어 이를 통해 구분하여 정확히 한번만 전달하게 할 수 있는 멱등성 프로듀서 동작 여부 설정

PID는 어떠한 이슈로 인해 애플리케이션이 재시작 됬을 경우, 이전 이슈로 전송되지 못했던 메시지가 같은 데이터라도 PID값은 새로 부여되어 달라진다.

해당 옵션 true 설정 시, retries옵션은 Integer.MAX, acks옵션은 all로 자동 설정된다.

• transactional.id: 프로듀서가 레코드 전송 시 레코드를 트랜잭션 단위로 묶을지 여부를 설정(default: null)

컨슈머는 기본적으로 파티션에 데이터가 쌓이는 족족 가져가지만, 트랜잭션으로 묶이게 되면 브로커에서 가져갈 때 다르게 동작하게 된다.

트랜색션 프로듀서를 사용하려면 enbale.idempotence=true로 설정하고, transaction.id는 임의의 String 값, 컨슈머에서는 isolation.level=read_committed로 설정하면 프로듀서와 컨슈머는 트랜잭션으로 처리 완료된 데이터만 읽고 쓰고 읽게 된다.

트랜잭션의 시작과 끝을 구분하기 위해 프로듀서에서 트랜잭션이 완료되었음만을 알리는 레코드를 마지막에 한번 더 전송하고 컨슈머는 파티션에 저장된 트랜잭션 레코드를 보고 트랜잭션이 완료되었다는 레코드를 확인했을 때 데이터를 전부 가져간다.

커스텀 파티셔너

특정 메시지 키를 가진 레코드는 0번 파티션에만 저장되도록 Partitioner 인터페이스를 직접 구현하여 커스텀하게 동작하는 파티셔너를 구현할 수 있다.

커스텀 파티셔너를 지정한 경우, ProducerConfig.PARTITION_CLASS_CONFIG 옵션을 사용자 생성 파티셔너로 설정하여 KafkaProducer 인스턴스를 생성해야 한다.

프로듀서의 브로커 정상 전송 여부 확인 처리

동기 처리

ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, message);
RecordMetadata metadata = producer.send(record).get();

KafkaProducer의 send() 메서드는 Future 객체를 반환하는 데 이 객체는 RecordMetadata의 비동기 결과를 표현하는 것으로 카프카 브로커에 정상적으로 저장이 되었는지에 대한 데이터가 포함되어 있다.

send()메서드에서 .get() 메서드를 사용하면 동기적으로 전송 결과 데이터를 알 수 있다.

동기로 프로듀서 전송 결과를 얻는 것은 성능에 제약을 준다.

비동기 처리

Callback 인터페이스를 구현하거나 send() 메서드가 반환한 ListenableFuture객체에 ListenableFutureCallback 인터페이스를 추가하고 onSuccess, onFailure 메서드를 오버라이드한 객체를 구현하여 레코드의 비동기 결과를 얻을 수 있다.

    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override 
        public void onSuccess(SendResult<String, String> result) {
            log.info("send success");
        }

        public void onFailure(Throwable ex) {
            log.error("send error! topic={}, message={}", topicName, message);
        }
    });

비동기 전송 결과는 처리 속도는 빠르지만 데이터의 순서가 중요한 경우에는 사용하면 안된다.