[Kafka] Kafka 클라이언트

💜 Kafka 클라이언트

책 '아파치 카프카 애플리케이션 프로그래밍 with 자바 - 최원영 지음'을 공부하며 정리하였습니다.

 

1. 카프카 클라이언트

카프카 클러스터 명령을 내리거나 데이터를 송수신하기 위해 카프카 클라이언트 라이브러리는 카프카 프로듀서, 컨슈머, 어드민 클라이언트를 제공하는 카프카 클라이언트를 사용하여 애플리케이션을 제공한다.
카프카 클라이언트는 라이브러리이기 때문에 자체 라이프사이클을 가진 프레임워크나 애플리케이션 위에서 구현하고 실행 해야한다.

---

(1) 프로듀서 API

프로듀서는 데이터 시작점으로, 프로듀서 애플리케이션은 카프카에 필요한 데이터를 선언하고 브로커의 특정 토픽의 파티션에 전송한다.
프로듀서는 데이터를 전송할 때 리더 파티션을 가지고 있는 카프카 브로커와 직접 통신한다.
프로듀서를 구현하는 가장 기초적인 방법은 카프카 클라이언트를 라이브러리로 추가하여 자바 기본 애플리케이션을 만드는 것이다.

(1)-1 프로듀서 내부 로직

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

 

[ Java 기준 프로듀서 애플리케이션 절차 ]

1. 전송하고자 하는 데이터를 ProducerRecord 클래스를 통해 인스턴스 생성하고, send() 메서드 호출
2. ProducerRecord는 파티셔너(partitioner)에서 토픽의 어느 파티션으로 전송될 것인지 정해진다.
   • KafkaProducer 인스턴스를 생성할 때 파티셔너를 따로 설정하지 않으면 DefaultPartioner로 설정된다.
3. 파티셔너에 의해 구분된 레코드는 데이터를 전송하기 전에 어큐뮬레이터(Accumulator)에 데이터를 버퍼로 쌓아놓고 발송한다.
   • 버퍼로 쌓인 데이터는 배치로 묶어서 전송된다. 이는 카프카 프로듀서 처리량을 향상시키는데 도움을 준다.
4. 센더(sender) 스레드는 Accumulator에 쌓인 배치 데이터를 가져가 파크파 브로커로 전송한다.

 

프로듀서 API 파티션 종류프로듀서 API는 2개의 파티션을 제공한다.

둘 다 메시지 키가 있을 때는 메시지 키의 해시값과 파티션을 매칭하여 데이터를 전송한다. 다만 메시지 키가 없을 때는 파티션에 최대한 동일하게 분배하는 로직 방식이 다르다.

• UniformStickyPartitioner :
  • 버전 2.4.0 이상에서 기본 설정되는 옵션으로, 프로듀서 동작에 특화되어 높은 처리량과 낮은 리소스 사용률을 가진다.
  • Accumulator에서 데이터가 배치로 모두 묶일 때까지 기다렸다가 배치로 묶인 데이터는 모두 동일한 파티션에 전송함으로써, RoundRobinPartitioner에 비해 향상된 성능을 가진다.
• RoundRobinPartitioner:
  • ProducerRecord가 들어오는 대로 파티션을 순회하면서 전송하기 때문에 배치로 묶이는 빈도가 적다.

 

---

(2) 컨슈머 API

컨슈머는 적재된 데이터를 사용하기 위해 브로커로부터 데이터를 가져와서 필요한 처리를 한다.

컨슈머를 구현하는 가장 기초적인 방법은 프로듀서와 동일하게 카프카 클라이언트를 라이브러리로 추가하여 자바 기본 애플리케이션을 만드는 것이다.

 

(2)-1 컨슈머 운영 방식

토픽의 파티션으로부터 데이터를 가져가기 위해 컨슈머를 운영하는 방법은 크게 2가지이다.

1. 1개 이상의 컨슈머로 이루어진 컨슈머 그룹을 운영
2. 토픽의 특정 파티션만 구독하는 컨슈머를 운영

컨슈머 그룹으로 묶인 컨슈머들은 토픽의 1개 이상의 파티션에 할당되어 데이터를 가져갈 수 있다.

• 1개의 파티션은 최대 1개의 컨슈머에 할당 가능하고, 1개의 컨슈머는 여러 개의 파티션에 할당될 수 있다. (컨슈머 그룹의 컨슈머 개수 <= 가져가고자 하는 토픽의 파티션 개수)
• 컨슈머 그룹은 다른 컨슈머 그룹과 격리되므로 그룹끼리 서로 영향을 주지 않는다.
  이러한 특징은 최종 적재되는 저장소의 장애에 유연하게 대응할 수 있도록 각기 다른 저장소에 저장하는 컨슈머를 다른 컨슈머 그룹으로 묶음으로써 각 저장소의 장애에 격리되어 운영할 수 있다.

 

(2)-2 컨슈머 장애 발생 대응 (리벨런싱)

리벨런싱

컨슈머 그룹의 특정 컨슈머에 장애가 발생하면, 장애가 발생한 컨슈머에게 할당된 파티션은 장애가 발생하지 않은 컨슈머에게 소유권이 넘어간다. ⇒ 리벨런싱(rebalancing)

리벨런싱이 발생하는 상황

1. 컨슈머가 추가되는 상황
2. 컨슈머가 제외되는 상황

컨슈머 중 1개에 이슈가 발생하여 동작을 안 하고 있다면 데이터 처리에 지연이 발생할 수 있다. 이를 해소하기 위해 이슈가 발생한 컨슈머를 컨슈머 그룹에서 제외하여 모든 파티션이 지속적으로 데이터를 처리할 수 있도록 가용성을 높여준다.

리벨런싱은 데이터 처리 도중 언제든 발생할 수 있으므로 리밸런싱에 대응하는 코드를 작성해야 한다.

그룹 조정자

그룹 조정자(group coordinator)는 리밸런싱을 발동시키는 역할을 하는데 컨슈머 그룹의 컨슈머가 추가되고 삭제될 때를 감지한다. 이는 카프카 브로커 중 한 대가 그룹 조정자 역할을 한다.

 

 

(2)-3 컨슈머 내부 구조

오프셋 커밋

컨슈머는 브로커로부터 데이터를 어디까지 가져갔는지 커밋(commit)을 통해 기록한다.
특정 토픽의 파티션을 어떤 컨슈머 그룹이 몇 번째 가져갔는지 브로커 내부에 사용되는 내부 토픽(_consumer_offsets)에 기록된다.
만약 컨슈머 동작 이슈로, 오프셋 커밋을 기록 못했다면 데이터 처리 중복이 발생할 수 있으므로, 컨슈머 애플리케이션이 오프셋 커밋을 정상적으로 처리했는지 검증해야만 한다.

 

오프셋 커밋은 컨슈머 애플리케이션에서 명시적, 비명시적으로 수행할 수 있다.

• 비명시 오프셋 커밋 : 일정 간격마다 자동으로 커밋 (enable.auto.commit=true)
  • poll() 메서드가 auto.commit.interval.ms에 설정된 값 이상이 지났을 때 그 시점까지 읽은 레코드의 오프셋을 커밋함
  • poll() 메서드 호출할 때 커밋을 수행하므로 코드상으로 커밋 관련 코드 따로 작성할 필요 없다. 다만, poll() 메서드 호출 이후에 리밸런싱 또는 컨슈머 강제종료 발생 시 컨슈머가 처리하는 데이터가 중복 또는 유실될 가능성이 있는 취약한 구조를 갖고 있다.
• 명시적 오프셋 커밋 : poll() 메서드 호출 이후에 반환받은 데이터의 처리가 완료되고 commitSync() 메서드를 호출
  • commitSync()는 poll()를 통해 반환된 레코드의 가장 마지막 오프셋을 기준으로 커밋을 수행함
  • commitSync()를 사용하여 커밋 요청을 전송하고 응답이 오기 전까지 데이터를 처리한다.

 

컨슈머 내부 구조

컨슈머는 자바 기준 poll() 메서드를 통해 레코드들을 반환받지만, poll() 메서드를 호출하는 시점에 클러스터에서 데이터를 가져오지 않는다.
[ 데이터 가져오는 과정 ]

1. 컨슈머 애플리케이션을 실행하면 내부에서 Fetcher 인스턴스가 생성됨
2. poll() 메서드를 호출하면 컨슈머는 내부 큐에 있는 레코드를 반환하여 처리를 수행

 

---

(3) 어드민 API

실제 운영환경에서 카프카 내부 옵션을 설정하고 확인하는 것이 중요하다.
내부 옵션을 확인하는 가장 확실한 방법은 브로커 한 대에 접소갛여 카프카 브로커 옵션을 확인하는 것이지만 번거롭다. 카프카 커맨드 라인 인터페이스로 명령을 내려 확인하는 방법도 있지만 일회성 작업이다.

 

카프카 클라이언트에서는 내부 옵션들을 설정하거나 조회하기 위해 AdminClient 클래스를 제공한다. 해당 클래스를 사용하면 클러스터 옵션과 관련된 부분을 자동화할 수 있다.

 

[ AdminClient 기능 ]

• 브로커 정보 조회
• 토픽 리스트 조회
• 컨슈머 그룹 조회
• 신규 토픽 생성
• 파티션 개수 변경
• 접근 제어 규칙 생성