Kafka Consumer(카프카 컨슈머)

컨슈머

---

컨슈머는 카프카의 3개 중요한 컴포넌트(프로듀서, 브로커, 컨슈머) 중 하나입니다.

컨슈머의 역활은 프로듀서에서 전송한 데이터가 브로커에 적재되게 되고 해당 데이터를 가져와서 처리하는 역활을 수행합니다.

 

컨슈머 운영방법

---

컨슈머를 운영하는 방법에는 크게 2가지가 있습니다.

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

 

1개 이상의 컨슈머로 이루어진 컨슈머 그룹을 운영

격리된 컨슈머 그룹을 이용한 데이터 전송 방법

 

컨슈머를 다른 컨슈머 컨슈머 그룹으로부터 격리된 환경에서 안전하게 운영할 수 있도록 도와주는 방식입니다.

 

이 방법을 사용할 때 고려해야할 점은 '컨슈머 개수는 토픽의 파티션의 개수보다 같거나 작아야 한다.'입니다.

왜냐하면 컨슈머의 개수가 파티션의 개수보다 많아지면 놀고 있는 파티션이 발생할 수 있기 때문입니다.

 

이 방법에 장점은 안전하게 운영이 가능하다는 것입니다.

엘라스틱 서치와 하둡을 사용하는 예를 들어 볼 수 있는데

엘라스틱 서치와 하둡을 사용하는 예시

 

위에 예시를 보게 되면 엘라스틱 서치 컨슈머에서 만약 문제가 발생하게 된다면 하둡 적재 컨슈머 그룹이 영향을 받을까?

그에 대한 답은 영향을 받지 않는다.입니다.

이게 1개 이상의 컨슈머로 이루어진 컨슈머 그룹을 운영하는 장점이라고 볼 수 있습니다.

 

이 장점을 볼 때 어디에 적재되고, 어떻게 처리되는지를 파악해서 컨슈머 그룹을 따로 나눌 수 있는 것은 최대한 나누는 것이 좋습니다.

 

그렇다면 컨슈머 그룹 내에서 장애가 발생한다면 카프카는 어떻게 처리하게 될까?

컨슈머에 장애가 발생하게 되면 장애가 발생된 컨슈머에 할당된 파티션을 장애가 발생되지 않은 컨슈머로 소유권을 넘겨 수행하게 됩니다.

이것을 리밸런싱이라고 부릅니다.

 

토픽의 특정 파티션만 구독하는 컨슈머를 운영하는 방법

이 방법은 웬만하면 사용하지 않는 것이 좋다고 생각합니다.

그 이유는 이 방법은 컨슈머가 assign() 메서드를 통해 직접 특정 토픽, 특정 파티션을 할당받아 사용하기 때문에

리밸런싱 하는 과정이 없어 장애가 발생하는 경우 대응하기가 쉽지 않을 것이라고 봅니다.

 

오프셋 커밋

---

컨슈머는 __consumer_offsets을 통해 데이터를 어디까지 가져갔는지 기록합니다. 

 

이것을 알려면 컨슈머 내부 동작에 대해서 알아보아야 할 것입니다.

출처 : https://blog.voidmainvoid.net/181 (Kafka Consumer 데이터 내부 처리 순서)

 

데브 원영님의 컨슈머 내부 처리 방식에 대한 이미지를 통해서 알아보려고 합니다.

1. poll() 메서드를 호출 Fetcher에 가져오고자 하는 레코드가 있는지 확인합니다.
2. 가져오고자 하는 레코드가 Fetcher 큐에 없는 경우, Fetch Request를 통해 브로커로부터 레코드를 가져옵니다.
3. 레코드가 있다면, 레코드 batch를 통해 Fetcher queue에 저장합니다.
4. 어디까지 읽었는지에 대한 offset을 내부 토픽과 컨슈머에 기록합니다.
5. 레코드 batch의 압축을 풀고, 레코드를 thread에 반환합니다.

내부 동작은 이렇게 끝나게 되는데, 여기서 추가할 점은 레코드를 오프셋 커밋함으로써 데이터가 안정적으로 처리됐음을 컨슈머 내부에 저장하고

먼저 설정한 retention ExpireTime에 따라 카프카 브로커가 ExpireTime이 지난 offset들을 컨슈머에 찾고 해당 레코드가 commit 됐다면 컨슈머 그룹에 대한 정보가 제외됨으로써 처리가 마무리 되게 됩니다.

만약 offset이 커밋되지 않았다면 데이터 중복이 발생할 수 있기 때문에 정상적으로 처리됐는지 검증할 필요가 있습니다.

 

오프셋을 커밋하는 방법은 명시적이거나 비명시적으로 할 수 있는데

방법은 enable.auto.commit = true이라면 명시적이고 false면 비명시적이다.

 

• 명시적 : poll() 메서드 호출 이후 commitSync() 메서드를 호출하면 됩니다.
   poll() 메서드를 통해 반환된 레코드의 가장 마지막 오프셋을 기준으로 커밋을 수행합니다.
• 비명시적 : 기본 옵션으로 poll() 메서드가 수행될 때 일정 간격마다 오프셋을 커밋합니다.
   auto.commit.interval.ms에 설정된 값으로 이 시간이 지났을 때 그 시점까지 읽은 레코드의 오프셋을 커밋합니다.
  주의할 점은 비명시적 오프셋 커밋은 편리하지만 리벨런싱 또는 컨슈머 강제종료 발생 시 데이터 중복 또는 유실이 발생될 수 있는 가능성이 있습니다.

 

출처 : 아파치 카프카 애플리케이션 프로그래밍 with 자바

아파치 카프카 애플리케이션 프로그래밍 with 자바 | 최원영 - 교보문고

 

아파치 Kafka Consumer의 데이터 처리 내부 architecture 설명 및 튜닝포인트

아파치 Kafka Consumer의 데이터 처리 내부 architecture 설명 및 튜닝포인트