Inflearn コミュニティ Q&A

프로듀서가 계속 데이터를 올리고 있는데도 컨슈머가 모든 파티션을 컨슘하고 있지 않다면 이 현상은 정상입니다.

캡쳐해주신대로 consumer 마다 파티션 지정은 완료된 상태입니다.

위에 올려주신대로 보면 컨슈머1은 파티션 0, 1, 2 담당하고 있고 컨슈머 2는 파티션 3,4,5 담당하고 있는데

컨슈머1: 파티션 1만 처리 중
컨슈머2: 파티션 3, 5 만 처리중
이런 상태이죠?

이건 컨슈머가 컨슘할 때 여러 파티션에서 메시지를 골고루 꺼내오지 않아서 그렇습니다.
poll_consumer에서 메시지를 한번에 100개씩 꺼내오도록 설정되어 있을 거에요. 바구니 크기가 100이라 해볼께요. 그럼 컨슈머는 특정 파티션 하나에서 우선 가져올 수 있는 만큼 가져옵니다. 그럼 특정 파티션 하나에서 이미 100개가 다 찹니다. 그래서 다른 파티션에서 가져올 여유가 없어서 지금 마치 하나의 파티션만 컨슘하고 있는 것처럼 보이게 됩니다.
(예: 컨슈머1: 파티션 1 / 컨슈머2: 파티션3 )

그러다가 컨슈머가 어느 순간 파티션을 바꾸어 가지고 오기도 합니다.

(예: 컨슈머1: 파티션 1 / 컨슈머2: 파티션5 )

지금 올려주신걸 보면 3개만 가져오는 걸로 보이는데 정확히는 파티션 3개를 동시 컨슘하고 있는게 아니라 여전히 파티션 2개를 컨슘하고 있는 중입니다. 3개로 보이는 이유는 commit 기록이 남아 있어서 보일 뿐입니다. (예: 파티션 3)

저 상황에서 한번 캡쳐 떠보시고, 한 10초 뒤에 캡쳐 다시 떠보시면 파티션 2개만 컨슈머의 current offset이 증가한게 보일거에요. 나머지 1개의 current offset은 그대로 일겁니다. 다만 예전에 찍어놓은 commit 기록이 있어서 보일 뿐이에요.

그래서 밤바미님이 지금 테스트하신게, Producer의 속도가 압도적으로 높은 상황에서 consumer가 따라가지 못하는 상황을 본 것입니다. 일반적으로 kafka 는 컨슈머의 부하량이 더 높습니다. 그래서 컨슈머의 부하 처리에 더 관심이 높고,

• 컨슈머 프로그램을 파티션 개수만큼 띄우는 이유

• 컨슈머 처리량으 느려지면 파티션을더 늘리고 컨슈머를 그만큼 더 띄우는 이유

모두 컨슈머가 속도를 못 따라갈때 대응하기 위한 방법들입니다. 그래서 컨슈머 6개를 올려보세요. 서버는 중복되어도 상관없습니다. 서버1에서 2개, 서버2에서 2개, 서버3에서 2개 올리시고

kafka web ui에서 consumer 의 컨슘 현황 캡쳐해보고, 10초 뒤에 다시 캡쳐해서 current_offset 비교해보시면 파티션 6개 처리되고 있는게 보일거에요.

그리고 consume_consumer.py 에서 로직을 이렇게 바꿔서 테스트해보세요.

                # 로직 처리 부분
                # Kafka 레코드에 대한 전처리, Target Sink 등 수행
                self.logger.info(f'message 처리 로직 시작')
                from collections import defaultdict
                partition_count = defaultdict(int)

                for msg in msg_lst:
                   if msg is None:
                      continue
                   if msg.error():
                      print("Error:", msg.error())
                      continue

                   partition_count[msg.partition()] += 1

                print("Partition message count:")
                for partition, count in partition_count.items():
                   print(f"partition {partition}: {count}")
                #msg_val_lst = [json.loads(msg.value().decode('utf-8')) for msg in msg_lst]
                #df = pd.DataFrame(msg_val_lst)
                #print(df[:10])

그럼 파티션별로 메시지를 몇 개 꺼내오는지 출력해서 확인할 수 있습니다.

구동해보시면 파티션 1개에서만 가져오는게 보일 거에요. 간혹 파티션 2개에서 꺼내오는 순간이 있긴한데 대부분 파티션 1개에서만 가져옵니다.

한번 테스트해보시겠어요?

안녕하세요 밤바미님!

먼저 늦게 답변드려 죄송합니다.

다양한 실험을 하고 계시는군요 ^^ 혹시 컨슈머를 추가하고나서 메시지는 계속 유입되고 있는 상태인가요? 아니면 메시지 유입은 중단된 상태인가요? 만약 메시지 유입이 중단된 상태라면 producer를 기동시켜 보시겠어요?

안녕하세요, 인프런 AI 인턴입니다. 밤바미님!

Kafka 컨슈머 그룹에서 특정 파티션이 읽히지 않는 문제는 여러 원인에 의해 발생할 수 있습니다. 이런 현상은 Kafka의 파티션 할당 방식 및 컨슈머의 구성과 관련이 있을 가능성이 큽니다. 다음 몇 가지 가능한 원인과 해결 방법을 고려해 보세요.

1. 파티션 개수 vs 컨슈머 개수: 컨슈머의 개수가 파티션의 개수와 일치해야 모든 파티션이 병렬로 읽히게 됩니다. 만약 컨슈머 수가 파티션 수보다 적으면 한 컨슈머가 여러 파티션을 읽게 되며, 모든 파티션을 커버하지 못할 수 있습니다.

2. 컨슈머 그룹 리밸런싱 문제: 컨슈머 그룹에 새로운 컨슈머가 추가되거나 제거될 때 컨슈머 그룹 리밸런스가 일어납니다. 이 과정에서 잠시 동안 파티션 할당이 어긋날 수 있으므로 상태를 지속적으로 모니터링해야 합니다.

3. 오프셋 설정 문제: 컨슈머가 특정 오프셋 이후로 데이터 읽기를 시작하도록 설정되어 문제가 발생할 수 있습니다. ‘auto.offset.reset’ 설정을 ‘earliest’로 바꿔 초기 오프셋에서부터 데이터를 읽도록 설정 변경을 고려해보세요.

4. 컨슈머 설정 확인: 각 컨슈머에 의해 올바르게 파티션이 할당되었는지, 할당 전략이 적절한지 확인해야 합니다. 대표적인 파티션 할당 전략은 ‘Range’와 ‘RoundRobin’ 방식이며, 특성에 따라 파티션이 고르게 할당되지 않을 수 있으므로 설정을 확인해 보세요.

5. 네트워크 및 리소스 문제: 충분한 네트워크 대역폭이 있지 않거나 컨슈머의 자원이 부족한 경우, 일부 파티션에 대한 메시지 처리 속도가 늦어질 수 있습니다. 컨슈머 실행 환경을 검토하고 필요하다면 리소스를 확장해 보세요.

이미지에서 볼 수 있는 다수의 파티션과 관련 컨슈머의 상태를 제공하고 있으니, 이상적인 파티션-컨슈머 매핑이 이루어졌는지 확인해보세요.

추가적으로, Kafka의 컨슈머와 파티션 관리에 대한 좀 더 깊이 있는 이해를 돕기 위해 이 설명서를 참고 하실 수 있습니다.

저는 질문자님의 이해를 돕고자 지식공유자님의 기존 답변을 참고하여 유사한 답변 링크를 추천해드리고 있습니다.

현재 베타 기능으로 답변이 만족스럽지 않을 수 있는 점 양해 부탁드립니다. 🙏
추가적으로 궁금한 점이 있으시면, 이어서 질문해 주세요. 곧 지식공유자께서 답변해 주실 것입니다.