[카프카 챌린지] 3부: Producer와 Record

애플리케이션에서 발생한 이벤트가 Kafka Broker에 저장되기까지는 Producer 내부의 여러 단계를 거친다.

Producer는 Kafka에 이벤트를 적재하는 역할을 하지만, send()를 호출한다고 해서 Record가 곧바로 Broker로 전송되는 것은 아니다. 내부적으로는 직렬화, 파티션 결정, 배치 누적, 전송 요청 생성 과정을 거친다.

이 글에서는 이벤트가 Record로 만들어진 뒤 Producer 내부에서 어떤 단계를 거쳐 Kafka Broker에 저장되는지 흐름 중심으로 살펴본다.

 

이벤트는 Record로 변환되어 전송된다

애플리케이션에서 발생한 이벤트가 그대로 Kafka에 저장되는 것은 아니다. Producer는 Kafka에 보낼 데이터를 Record 형태로 만들어 전송한다.

Record에는 이벤트가 저장될 Topic, 실제 저장할 내용인 Value, 그리고 필요하다면 Partition을 결정하는 데 사용되는 Key 같은 정보가 담긴다. 대표적으로 자주 사용되는 필드는 다음과 같다.

topic	이벤트가 저장될 Topic
key	Partition을 결정하는 기준이 될 수 있는 값
value	실제로 저장하려는 이벤트 내용
partition	직접 저장할 Partition 번호를 지정할 때 사용
headers	추적 ID, 스키마 버전 같은 부가 메타데이터

 

 

예를 들어 서버 에러 로그를 Kafka에 저장한다면 다음과 같이 생각할 수 있다.

topic = error_logs
key = server-1
value = "fatal error occurred"

 

여기서 topic은 이벤트를 저장할 카테고리이고, value는 실제 저장할 내용이다. key는 이 Record가 어느 Partition으로 갈지 결정할 때 사용될 수 있다.

이렇게 만들어진 Record를 Producer에게 넘기는 지점이 바로 send() 호출이다.

 

send()는 바로 전송을 의미하지 않는다

ProducerRecord를 만들었다면 이제 Producer의 send()를 호출해 Kafka로 전송을 요청한다. 이름만 보면 send()를 호출하는 순간 Record가 바로 Broker로 전송될 것 같지만, 실제로는 그렇지 않다.

 

send()는 Record를 곧바로 Broker에 쓰는 동작이라기보다, Producer 내부 전송 흐름을 시작하는 진입점에 가깝다. 애플리케이션은 Producer에게 전송을 요청하고, Producer는 내부에서 필요한 처리를 거친 뒤 적절한 시점에 Broker로 전송한다.

 

send() 이후에는 먼저 Record의 Key와 Value가 Kafka가 다룰 수 있는 형태로 변환된다. 그 다음 이 Record가 어느 Partition으로 가야 하는지 결정되고, 바로 Broker로 보내지는 대신 내부 버퍼에 Batch 단위로 모인다.

 

따라서 send()를 호출했다는 것은 "Record를 Broker에 바로 저장했다"는 의미가 아니라, Producer에게 이 Record를 Kafka로 전송해 달라고 맡긴 것에 가깝다.

 

Serialization: Kafka가 다룰 수 있는 형태로 바꾸기

send()로 전달된 Record는 먼저 Serialization 과정을 거친다. Serialization은 애플리케이션에서 사용하는 문자열, 숫자, 객체 같은 데이터를 Kafka가 저장하고 전송할 수 있는 바이트 형태로 바꾸는 과정이다.

앞에서 예로 든 Record의 key와 value는 애플리케이션 코드에서는 문자열처럼 보인다.

key = "server-1"
value = "fatal error occurred"

 

하지만 Kafka는 이 값을 문자열 그대로 저장하는 것이 아니라, 바이트 데이터로 변환된 결과를 저장하고 전송한다.

key → serialized key
value → serialized value

 

이 변환은 Serializer가 담당한다. Java 기준으로 String 값을 보낸다면 StringSerializer를 사용할 수 있고, 직접 만든 객체를 보낸다면 JSON, Avro, Protobuf 같은 직렬화 방식을 사용할 수 있다.

Record가 Kafka가 다룰 수 있는 바이트 형태로 바뀌었다면, 다음으로는 이 Record가 어느 Partition에 저장될지 결정해야 한다.

 

Partitioning: 어느 Partition으로 보낼지 결정하기

직렬화가 끝나면 Producer는 이 Record를 어느 Partition으로 보낼지 결정한다. 이 과정을 Partitioning이라고 한다.

여기서 말하는 Partitioning은 Topic을 여러 Partition으로 나누는 과정이 아니라, 현재 전송하려는 Record가 몇 번 Partition에 저장될지 결정하는 과정이다.

 

Partition은 대략 다음 순서로 결정된다.

Partition을 직접 지정했다면 해당 Partition 사용
→ 직접 지정하지 않았다면 Custom Partitioner 확인
→ Key가 있다면 Key를 Hashing해서 Partition 결정
→ Key가 없다면 기본 Partitioning 방식 사용

 

일반적으로는 Partition을 직접 지정하기보다 key를 기준으로 Partition이 결정되는 경우가 많다. 같은 key는 같은 Hashing 결과를 만들기 때문에, 보통 같은 Partition으로 이동하게 된다.

 

앞에서 본 서버 에러 로그 예시에서는 server-1을 key로 사용했다. 이 경우 server-1에서 발생한 로그들은 같은 Partition에 저장될 수 있다.

 

Kafka는 Topic 전체의 순서를 보장하는 것이 아니라 Partition 내부의 순서를 보장한다. 따라서 서버별 로그 순서가 중요하다면 serverId를 key로 사용하고, 주문 상태 변경 순서가 중요하다면 orderId를 key로 사용하는 식으로 순서를 유지하고 싶은 기준을 key로 잡는 것이 중요하다.

 

Partitioning은 단순히 Record를 아무 Partition에나 나눠 넣는 과정이 아니라, 데이터를 어떤 기준으로 분산하고 같은 Partition에 묶을지 결정하는 과정이다.

이제 Record가 저장될 목적지는 정해졌다. 하지만 목적지가 정해졌다고 해서 바로 네트워크로 전송되는 것은 아니다.

 

Accumulation: 바로 보내지 않고 Batch로 모으기

Partition까지 결정되면 이제 Broker로 바로 전송될 것 같지만, Producer는 Record를 한 번 더 내부에 모아둔다. 이때 사용되는 공간이 RecordAccumulator다. RecordAccumulator는 Kafka Broker에 실제로 저장되는 공간이 아니라, Broker로 보내기 전에 Producer 내부에서 대기하는 전송 버퍼에 가깝다.

 

이때 Record는 아무렇게나 섞여서 저장되는 것이 아니라, Topic과 Partition별로 구분되어 쌓인다. 예를 들어 error_logs Topic에 Partition이 2개 있다면, Producer 내부에서는 error_logs-0으로 갈 Record와 error_logs-1로 갈 Record가 각각 따로 모인다.

RecordAccumulator

error_logs-0 → Batch
error_logs-1 → Batch

 

Producer가 이렇게 Record를 모아두는 이유는 여러 Record를 하나의 Batch로 묶어 전송하기 위해서다. Record 하나가 생길 때마다 매번 네트워크 요청을 보내면 요청 횟수가 많아지고, 요청 생성과 응답 처리 같은 고정 비용도 계속 반복된다.

 

이렇게 Accumulator에 모인 Batch는 Producer 내부의 백그라운드 Thread에 의해 실제 전송 단계로 넘어간다.

 

Sender Thread가 실제 전송을 담당한다

Producer 내부에 Batch가 쌓였다고 해서 애플리케이션 코드가 직접 Broker로 전송하는 것은 아니다. 실제 전송은 Producer 내부에서 백그라운드로 동작하는 Sender Thread가 담당한다.

 

애플리케이션 쪽에서는 send()를 통해 전송을 요청하고, Producer 내부에서는 Record를 Accumulator에 모아둔다. 이후 Sender Thread는 Accumulator에 쌓인 Batch 중 전송 가능한 것들을 꺼내 Kafka Broker로 보낼 요청을 만든다.

 

이 구조 덕분에 애플리케이션은 Record를 보낼 때마다 네트워크 전송이 끝날 때까지 기다릴 필요가 없다. 애플리케이션은 전송 요청을 맡기고, Producer 내부에서는 Batch를 모아 백그라운드에서 전송을 처리한다.

 

다만 Sender Thread가 Batch를 꺼냈다고 해서 각 Batch를 무조건 따로 보내는 것은 아니다. 실제 전송 요청은 Broker 단위로 다시 묶인다.

 

Broker별 ProduceRequest로 묶어 전송한다

Kafka에서는 각 Partition마다 데이터를 받을 수 있는 Leader Broker가 정해져 있다. 따라서 Sender Thread는 Batch를 꺼낸 뒤, Batch들이 어느 Broker로 가야 하는지 먼저 확인한다.

 

예를 들어 user_updates-0의 Leader가 Broker 0에 있고, error_logs-1의 Leader도 Broker 0에 있다면 두 Batch는 같은 Broker로 전송되어야 한다. 이 경우 Sender Thread는 같은 Broker로 향하는 Batch들을 하나의 ProduceRequest 안에 묶는다.

Broker 0으로 보낼 ProduceRequest
- user_updates-0 Batch
- error_logs-1 Batch

Broker 1로 보낼 ProduceRequest
- user_updates-1 Batch

 

이렇게 Broker 단위로 요청을 묶으면 네트워크 요청 횟수를 줄일 수 있다. Record를 하나씩 보내는 것보다 Batch로 묶는 것이 효율적이었던 것처럼, Broker로 보낼 때도 같은 목적지로 가는 데이터는 함께 묶어 보내는 편이 효율적이다.

 

Producer 내부의 흐름을 보면, Accumulator에서는 Topic-Partition 단위로 Record를 모으고, Sender Thread는 전송 단계에서 이를 다시 Broker 단위의 요청으로 재구성한다. 이 과정을 거쳐 만들어진 요청이 실제 Broker로 전송되는 ProduceRequest다.

 

ProduceRequest가 만들어지면 이제 실제 네트워크 전송 단계로 넘어간다.

 

NetworkClient와 InFlightRequests

Broker별로 만들어진 ProduceRequest는 NetworkClient를 통해 실제 Broker로 전송된다. Sender Thread가 전송할 요청을 만들었다면, NetworkClient는 그 요청을 네트워크를 통해 Broker에 보내는 역할을 한다.

 

전송이 끝났다고 해서 요청 처리가 바로 완료되는 것은 아니다. Producer는 Broker로부터 성공 또는 실패 응답을 받아야 한다. 그래서 NetworkClient 내부에서는 이미 보냈지만 아직 응답을 받지 못한 요청들을 따로 관리한다. 이때 사용되는 구조가 InFlightRequests다.

 

InFlightRequests는 이름 그대로 전송 중인 요청들을 의미한다. 예를 들어 Broker 0으로 요청을 보냈지만 아직 응답이 오지 않았다면, 해당 요청은 Broker 0의 InFlightRequests에 남아 있게 된다.

broker0 → [request1, request2]
broker1 → [request3]
broker2 → []

 

이 구조가 필요한 이유는 Producer가 여러 Broker와 동시에 통신하기 때문이다. 어떤 Broker에 어떤 요청을 보냈는지 알고 있어야 응답이 돌아왔을 때 올바른 요청과 매칭할 수 있고, 실패나 timeout이 발생했을 때도 재시도 여부를 판단할 수 있다.

 

Producer 입장에서 보면 ProduceRequest는 Broker로 보낼 요청이고, NetworkClient는 그 요청을 실제로 전송하는 역할을 한다. InFlightRequests는 전송했지만 아직 응답을 받지 못한 요청들을 관리하는 공간이다.

 

 

정리하기

Producer의 전송 과정은 단순히 send() 한 번으로 끝나지 않는다. 애플리케이션에서 발생한 이벤트는 먼저 Record로 만들어지고, Kafka가 다룰 수 있는 형태로 직렬화된다. 이후 어느 Partition으로 갈지 결정된 다음, Producer 내부의 RecordAccumulator에 Batch 단위로 모인다.

 

이벤트 발생
→ Record 생성
→ send 호출
→ Serialization
→ Partitioning
→ Accumulation
→ Sender Thread
→ ProduceRequest 생성
→ NetworkClient 전송
→ Broker 저장

 

이 흐름을 따라가면 Producer의 동작을 단순히 "Kafka에 메시지를 보낸다"가 아니라, "Record를 만들고, 직렬화하고, Partition을 결정하고, Batch로 모은 뒤, Broker별 요청으로 전송한다"는 과정으로 이해할 수 있다.

 

---

 

본 글은 쉬운코드 유튜브 채널의 회원 전용 Kafka 챌린지 라이브 강의를 기반으로 개인 학습 및 복습 목적으로 정리한 내용입니다.

원본 영상은 아래 링크에서 확인할 수 있습니다.

https://www.youtube.com/live/9d9aVj4rix4