Kafka 4

공부한 개념을 한 문장씩 쭉 세워본다.

 

카프카에서 데이터의 시작점은 프로듀서이다.

 

프로듀서는 데이터를 전송할 때 리더 파티션을 가지고 있는 카프카 브로커와 직접 통신한다.

 

카프카 공식 오픈 라이브러리 (Java) 기준으로..

ProducerRecord 를 KafkaProducer 의 send 메서드로 보낼 수 있다.

 

KafkaProducer 의 send 메서드를 호출 하면,

프로듀서 내부적으로 Partitioner, Accumulator, Sender 가 동작하고

실질적으로 데이터가 리더 파티션을 가진 카프카 브로커로 전송된다.

 

ProducerRecord 객체는 프로듀서에서 생성하는 레코드이다. (오프셋은 미포함)

KafkaProducer 의 send 메서드는 레코드를 전송하기 위한 요청 메서드이다.

Partitioner 는 어느 파티션(리더)으로 전송할지 지정하는 객체이다.

Accumulator 는 특정기간 동안 레코드를 모으는(배치 성격) 버퍼이다.

Sender 는 모인 레코드를 실질적으로 카프카로 보낸다.

 

ProducerRecord 생성 시, 토픽 이름과 메시지 값은 필수이다.

send 메서드의 반환 타입은 Future 이다.

Accumulator 의 존재로 인해 커넥션을 최적화 할 수 있다. (높은 데이터 처리량)

 

카프카 공식 오픈 라이브러리 (Java) 기준으로..

프로듀서의 파티셔너 구현체로..

UniformStickyPartitioner (기본) 와 RoundRobinPartitioner 를 제공한다.

 

메시지(레코드) 키가 존재하면,

두개의 Partitioner 모두 키를 해싱하여 동일한 파티션(리더) 에 매칭되도록 한다.

 

운영 중에 파티션 개수가 변경되면 키와 파티션 번호 매칭이 깨지므로 처음 부터 충분히 많은 파티션을 만들도록 하자.

 

메시지 키가 없으면(null),

두 파티셔너 모두 파티션을 순회하면서 레코드를 고르게 분배 시켜 전송시킨다.

 

UniformStickyPartitioner 는 RoundRobinPartitioner 와 비교하여 Accumulator 에서 묶이는 정도가 다르다.

좀 더 효율적으로 설계되어있다. (높은 데이터 처리량)

 

프로듀서에는..

Partitioner 를 상속받아 커스텀 Partitioner 개발하여 등록해 줄 수 있다.

예를 들면, key 와 value 를 조합하여 특정 파티션 번호에 지정할 수 있는 로직을 짤 수 있다.

 

KafkaProducer 를 생성할 때의 옵션

- bootstrap.servers 는 프로듀서가 데이터를 전송할 대상(브로커)의 호스트 이름이다. (2개 이상의 브로커를 입력하면 안전)

- key.serializer 는 메시지 키를 직렬화하는 클래스 지정

- value.serializer 는 메시지 값을 직렬화하는 클래스 지정

 

위 3개의 옵션은 필수이다.

 

직렬화 클래스는 토픽단위나 전역으로 설정가능하다.

 

직렬화 클래스는 보통 StringSerializer 로 지정한다.

그렇지 않으면 코드를 따로 관리해야하는 공수가 발생할 수 있다.

 

StringSerializer 로 지정하면, kafka-console-consumer 로 카프카에서 메시지를 확인 할 수 있다.

 

- acks 는 데이터 전송 신뢰도 정도에 대한 옵션이다.

프로듀서가 매칭된 리더 파티션을 가진 브로커에 데이터를 전송하면...

프로듀서가 전송한 데이터가 브로커들에 정상적으로 저장되었는지에 대한 확인 정도이다. (기본 값은 1이다.)

 

0 이면, 리더 파티션에 데이터가 저장되었는지 확인하지 않고 단순히 전송만 하는 옵션이다.

저장여부를 확인하지 않기 때문에 오프셋 값도 응답되지 않는다.

보통 GPS 데이터와 같은 도메인에 사용된다.

 

1 이면, 리더 파티션에 데이터가 저장되었는지 확인하고 응답된다.

저장되지 않았다면 재시도를 한다. (아래 retries 옵션과 관련)

리더 파티션에 데이터가 적재되었지만, 팔로워 파티션에서 데이터를 동기화를 아직 못했을 때

리더 파티션에 장애가 나면 데이터 유실이 이루어질 수 있다.

 

2(all) 이면, 리더 파티션 뿐만아니라 팔로워 파티션에도 데이터가 저장되었는지 확인하고 응답된다.

이 경우, ISR 그룹내의 몇개의 팔로워 파티션에 저장되었는지를 확인할 옵션이 있어야한다.

 

min.insync.replicas 옵션은

프로듀서가 리더, 팔로워 파티션에 데이터가 정상 적재되었는지 확인하기 위한 최소 ISR 그룹 파티션 개수이다.

 

예를 들어..

acks 가 2, replication factor 가 3, min.insync.replicas 가 2이면..

리더 외에 1개의 팔로워에 정상적재 됨을 확인하고 프로듀서에 응답을 보낸다.

 

acks 가 2, min.insync.replicas 가 1이면..

ISR 그룹내의 파티션은 리더 파티션도 포함이므로

리더 파티션에 정상 적재 여부만 확인하고 응답을 보낸다.

따라서, acks 가 2이지만 1과 다를 바가 없이 의미가 없어진다.

acks 가 2 이면, min.insync.replicas 가 2 이상이어야 의미가 있다.

 

acks 는 프로듀서 옵션이다.

min.insync.replicas 는 토픽 옵션이다.

 

- linger.ms 옵션은 Accumulator 가 얼마나 데이터를 모아서 전송할 것인지에 대한 시간 값이다. (기본 값은 0 이다.)

- retries 옵션은 브로커로 부터 에러를 받고 재전송을 시도하는 횟수이다. (기본 값은 INT_MAX 이다.)

- max.in.flight.requests.per.connection 옵션은

실제 데이터를 보내는 Sender 가 가지는 최대 커넥션 수이다. (기본 값은 5 이다.)

- partitioner.class 옵션은 KafkaProducer 를 생성할 때 지정하는 파티셔너 객체이다.

(기본 값은 org.apache.kafka.clients.producer.internals.DefaultPartitioner 이다.)

- enable.idempotence 옵션은 멱등성 프로듀서로 동작할지 여부이다. (기본 값은 false 이다.)

- transactional.id 옵션은 레코드를 트랜잭션 단위로 묶을지 여부이다. (기본 값은 null 이다.)