메시지 전송 절차#

1. ProducerRecord 객체 생성

객체 생성시 직렬화

• 필수 지정 : Topic, Value
• 선택 지정 : Key, Partition, Timestamp

2. 데이터를 파티셔너로 전송

• 파티셔너는 토픽의 파티션을 선택한다. 일반적으로 메시지의 Key를 기준으로 메시지를 선택한다.
• 전송 성공시 metadata 객체 반환

3. 레코드 배치에 추가

• 별도의 스레드가 레코드 배치를 적절한 카프카 브로커에게 전송

4. 카프카 브로커에 전송

5. 카프카 브로커 응답 반환

프로듀서 생성시 옵션들#

• bootstrap.servers : 브로커 목록을 지정한다.
• key.serializer : 메시지 키의 직렬화 방법을 지정한다.
• value.serializer : 메시지 값의 직렬화 방법을 지정한다.

프로듀서 인터페이스는 임의 자바 객체를 키 혹은 밸류로 전송할 수 있도록 매개변수화된 타입을 사용할 수 있도록 한다

프로듀서 설정을 위한 몇가지 예제들

• Spring Boot환경에서 yaml file에 프로듀서 설정 작성
• kafkaTemplate<> 사용
• @Configuration class 생성
• Properties()

메시지 전송 방법#

Concept

• Fire and Forget : 메시지를 서버에 전송만하고 성공 혹인 실패 여부에는 신경쓰지 않음. 재시도할 수 없는 에러 혹은 타임아웃 발생시 메시지가 유실됨.
• 동기적 전송 : 기본적으로 카프카는 비동기적으로 동작하지만(Future 객체를 반환) 다음 메시지를 전송하기 전 get()을 호출하여 동기적으로 동작하게끔 할 수 있다.
• 비동기적 전송 : 콜백함수와 함께 send() 를 호출하면 카프카 브로커로부터 응답을 받는 시점에서 자동으로 콜백 함수가 호출된다.

프로듀서 설정#

Concept

• client.id : 프로듀서 클라이언트 식별자. 브로커에 로그를 남길 때 사용.
• acks : 프로듀서 매개변수. 프로듀서가 임의의 쓰기 작업이 성공했다고 판별하기 위해 얼마나 많은 파티션 레플리카가 해당 메시지를 수신해야 하는지를 지정.
• end to end latency : 종단지연. 레코드가 생성되어 컨슈머가 읽을 수 있을 때까지의 시간. acks는 종단지연에 영향을 주지 않는다.

프로듀서 설정 예제

Spring Configuration 방식으로 설정

1
@Configuration
2
public class KafkaProducerConfig {
3

4
    // <String, String>
5
    // <String, customDTO>
6
    @Bean
7
    public ProducerFactory<String, Customer> factory() {
8

9
        return new DefaultKafkaProducerFactory<>(producerConfigs());
10
    }
11

12
    @Bean
13
    public KafkaTemplate<String, Customer> kafkaTemplate() {
14
        return new KafkaTemplate<>(factory());
15
    }
16

17
    @Bean
18
    public Map<String, Object> producerConfigs() {
19
        Map<String, Object> props = new HashMap<>();
20
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9095");
21
        props.put(ProducerConfig.ACKS_CONFIG, "all");
22
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
23
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
24
        props.put("schema.registry.url", "http://localhost:8084");
25

26
        return props;
27
    }
28

29
}

메시지 전달 시간 설정#

ProducerRecord를 보낼 때 걸리는 시간 두 구간

• send에 대한 비동기 호출이 이뤄진 시간부터 결과를 리턴할 때 까지 걸리는 시간.
• send에 대한 비동기호출이 리턴한 시각부터 콜백이 호출 될 때까지 걸리는 시간.

batch.size 설정#

• batch.size : 프로듀서가 각각의 메시지 배치에 사용될 메모리 양을 지정한다.
• batch.size 를 너무 작게 설정할 경우 작은 배치를 자주 전송하게 되어 약간의 오버헤드가 발생할 수 있다.

순서 보장하기#

• max.inflight.requests.per.connection : 프로듀서가 서버로부터 응답을 받지 못한 상태에서 전송할 수 있는 최대 메시지 의 수를 지정한다. 이 값을 올릴 경우 메모리 사용량이 증가하지만 처리량 또한 증가한다.

커스텀 직렬화#

커스텀 직렬화는 일반적으로 권장되지 않는다. 이는 기존 스키마와 신규 스키마 간의 호환성을 유지하기 어렵게 만들기 때문이다.

Avro 직렬화#

스키마 변경시 시나리오

• 데이터를 읽고 쓸때 사용되는 스키마가 호환되어야 한다.
• 역직렬화시 데이터를 쓸 대 사용했던 스키마에 접근이 가능해야 한다.

Sticky Partitioner#

접착성 처리

접착성 처리는 기본적으로 프로듀서가 메시지 배치를 채울 때, 다음 배치로 넘어가기 전에 이전 배치를 모두 채우는 것을 의미한다. 키 값이 null 인 메시지를 접착성 처리할 경우 일단 키 값이 있는 메시지에 따라 붙어야 RR 방식오로 배치되게 된다.

Partitioner#

Kafka에서 제공되는 파티셔너

Concept

• Sticky Partitioning : 프로듀서가 메시지 배치를 채울 때, 다음 배치로 넘어가기 전에 이전 배치를 모두 채우는 것
• RoundRobin Partitioner : 기본적으로 메시지를 전체 파티션에 대해 균등하게 할당하는 것
• UniformSticky Partitioner : batch.size에 도달할 때까지 파티션을 유지. 대기 중인 큐의 크기를 고려하여 파티션을 전환하는 균등 분배 전략 사용.

쿼터 타입#

• 쓰기 쿼터(produce quota)
• 읽기 쿼터(consume quota)
• 요청 쿼터(request quota)

리소스 관리하기#

쿼터, 쓰로틀링이 필요한 이유는 기본적으로 브로커의 리소스(CPU, Memory, Network) 를 안정적으로 관리하기 위함이다. 프로듀서가 브로커에 너무 많은 요청을 보낸 상태에서 브로커가 밀린 메시지를 처리하지 못해 프로듀서 버퍼의 메모리 공간이 확보되지 못할 경우 TimeoutException이 발생할 수 있다.

Key Takeaway

• Latency와 메시지 신뢰성의 Tradeoff를 고려해 프로듀서 설정을 조정할 수 있다.
• Latency와 Throughput의 Tradeoff를 고려해 프로듀서 설정을 조정할 수 있다.
• 비동기 호출이 시행되고 반환되는 시간과 콜백이 호출되는 시간을 나눠서 관제함으로써 메시지 전송에서의 병목 지점을 파악할 수 있다.
• 기본 제공타입 직렬화나 커스텀 직렬화 대신 Avro와 같은 직렬화 포맷을 활용하는 것이 권장된다.
• Schema Registry 를 통해 스키마 버전관리와 호환성을 유지할 수 있다.
• Sticky Partitioning 을 통해 더 적은 요청으로 같은 수의 메시지를 전송함으로써 Latency와 CPU 사용량을 줄일 수 있다.
• 파티션을 결정하는데 사용하는 키가 중요해서 같은 키값을 사용하는 파티션이 변경되어서는 안될 경우. 충분한 수의 파티션을 가진 토픽을 생성하고 그 이상 파티션을 추가하지 않는 것이 좋다.
• 카프카 레코드의 Key, Value를 건드리지 않고 추가 메타데이터를 작성할 경우 Header 를 사용한다.
• 쿼터, 쓰로틀링이 필요한 이유는 기본적으로 브로커의 리소스(CPU,Memory,Network) 를 안정적으로 관리하기 위함이다.