04. 카프카 프로듀서(Kafka Producer)

  프로듀서는 카프카에 데이터를 저장하는 첫 단계이자 시작점입니다. 카프카에 필요한 데이터를 선언하고 브로커의 특정 토픽의 파티션에 전송합니다. 프로듀서는 데이터를 전송할 때 리더 파티션을 가지고 있는 카프카 브로커와 직접 통신합니다. 프로듀서는 데이터를 직렬화하여 카프카 브로커로 보내기 때문에 자바에서 선언 가능한 모든 형태를 브로커로 전송할 수 있습니다. 직렬화란 자바 또는 외부 시스템에서 사용 가능하도록 바이트 형태로 데이터를 변환하는 기술입니다. 직렬화를 사용하면 프로듀서는 자바 기본형과 참조형뿐만 아니라 동영상, 이미지 같은 바이너리 데이터도 프로듀서를 통해 전송할 수 있습니다.

 

1. acks 옵션

  카프카 프로듀서의 acks옵션은 0, 1, all(or -1) 값을 가질 수 있습니다. 이 옵션을 통해 프로듀서가 전송한 데이터가 카프카 클러스터에 얼마나 신뢰성 높게 저장할지 지정할 수 있습니다. 복제 개수가 1인 경우 acks옵션에 따른 성능 변화는 크지 않지만, 대부분 안정적으로 데이터를 운영하기 위해 복제 개수를 2 이상으로 설정합니다.

 

acks=0

  프로듀서가 리더 파티션으로 데이터를 전송했을 때 리더 파티션으로 데이터가 저장되었는지 확인하지 않습니다. 프로듀서는 전송을 하자마자 데이터가 저장되었음을 가정하고 다음 데이터를 전송하기 때문에 retries 옵션값은 무의미합니다. 데이터가 일부 유실이 발생하더라도 전송 속도가 중요한 경우에 사용합니다.

 

acks=1

  프로듀서는 보낸 데이터가 리더 파티션에만 정상적으로 적재되었는지를 확인합니다. 리더 파티션에 적재될 때까지 재시도할 수 있습니다. 그러나 팔로워 파티션이 데이터를 복제하기 직전에 리더 파티션이 있는 브로커에 장애가 발생하면 동기화되지 못한 일부 데이터가 유실될 수 있습니다.

 

acks=all or acks=-1

  프로듀서는 보낸 데이터가 리더 파티션과 팔로워 파티션에 모두 정상적으로 적재되었는지 확인합니다. 0 또는 1 옵션보다 속도는 느리지만 안전하게 데이터를 전송하고 저장합니다. acks를 all로 설정할 경우 토픽 단위로 설정 가능한 min.insync.replicas 옵션 값에 따라 데이터의 안정성이 달라집니다.

  복제 개수를 3으로 설정하고 min.insync.replicas를 3으로 설정한 경우, 브로커 3대 중 1대에 이슈가 발생하여 동작하지 못하면 NotEnoughReplicasException or NotEnoughReplicasAfterAppendException이 발생합니다. 그러므로 토픽별 min.insync.replicas 옵션 값은 브로커 개수 미만으로 설정해서 운영해야 합니다.

 

2. 멱등성(idempotence) 프로듀서

  기본 프로듀서의 동작 방식은 적어도 한 번 전달(at least once delivery)을 지원합니다. 멱등성 프로듀서는 동일한 데이터를 여러 번 전송하더라도 카프카 클러스터에 단 한 번만 저장됨을 의미합니다. 단, 멱등성 프로듀서는 동일한 세션(PID의 생명주기)에서만 정확히 한 번 전달을 보장합니다. 아래 옵션 값을 설정해서 멱등성 프로듀서로 동작하도록 만듭니다. 아래 옵션 설정 시 retries는 기본값으로 Integer.MAX_VALUE, acks 옵션은 all로 강제 설정됩니다.

  enable.idempotence = true # (default : false)

  멱등성 프로듀서는 데이터를 브로커로 전달할 때 프로듀서 PID(Producer unique ID)와 시퀀스 넘버(sequence number)를 함께 전달합니다. 브로커에서 전송한 데이터의 PID와 시퀀스 넘버를 확인하는 과정에서 시퀀스 넘버가 일정하지 않은 경우 OutOfOrderSequenceException이 발생할 있습니다.

 

3. 트랜잭션(transaction) 프로듀서

  카프카의 트랜잭션 프로듀서는 다수의 파티션에 데이터를 저장할 경우 모든 데이터에 대해 동일한 원자성(atomic)을 만족시키기 위해 사용됩니다. 트랜잭션 프로듀서는 트랜잭션의 시작과 끝을 표현하기 위해 트랜잭션 레코드를 한 개 더 보냅니다. 트랜잭션 컨슈머는 파티션에 저장된 트랜잭션 레코드를 보고 트랜잭션이 완료(commit) 되었음을 확인하고 데이터를 가져갑니다. 설정 방법은 아래와 같습니다.

# Producer
  enable.idempotence = true && transactional.id = 임의의 String 값

# Consumer
  isolation.level = read_committed