잼의 이것저것

목차 Kafka의 에러 핸들링은 보통 Consumer단에서 처리한다고 한다. 이러한 에러를 핸들링하는 패턴들에 대해 정리. [Poison Pill] Poison Pill이란 Consumer가 처리하지 못하는 메시지가 왔을 때를 의미힌다. (Deserializer를 하지 못하는 메시지 형태) M2가 비정상적 메시지라고 하면 M2를 처리하다가 오류가 나고 Consumer가 실패할 것이다. Spring - ErrorHandlingDeserializer 스프링에서는 ErrorHandling Deserializer를 제공하고 있다고 한다. 그래서 이상한 데이터가 들어오면 ErrorHandling Logic의해 처리되고, 정상적인 데이터들만 Consumer로 넘어가게 하는 패턴을 권장한다고 한다. [Error Ha..

목차 [Consumer 관련 용어들] Log end Offset: 파티션에 추가된 최신 데이터. producer가 현재 쓰고 있는 위치. 아직은 consumer가 액세스 할 수 없음 High Watermark: 데이터가 모든 Replica에 복제됨. 사용가능한 오프셋 위치 Current Position: consumer instance 개별적으로 관리하는 것. 현재 처리 중인 오프셋 위치 Last Committed Offset: __consumer_offsets 토픽에 저장된 마지막 위치. [Conusmer 주요 파라미터들] fetch.min.bytes: 최소 fetch 용량. 1이면 즉시 불러온다. fetch.wait.max.ms: fetch 최대 대기시간. max.partition.fetch.byte..

목차 [Simple Consumer?] offset과 같은 세부적인 부분을 직접 다루고 싶은 요구가 있는 경우 Consumer 관련항목을 직접 제어하기 위해 사용. 이름은 Simple 이지만 실제로는 전혀 Simple하지 않음 Consumer-Group이라는 개념이 Simple Consumer에서는 없다. [Simple Consumer를 사용하는 이유] SimpleConsumer 구현을 사용하는 주된 이유는 Consumer-group이 제공하는 것보다 파티션 소비를 더 잘 제어하기를 원하기 때문. 예: 메시지를 여러 번 읽기 토픽의 일부 파티션만 사용하도록 제어 메시지가 한 번만 처리되도록 트랜잭션을 관리합니다. [Simple Consumer의 단점] SimpleConsumer는 Consumer Grou..

Confluent 문서에 리밸런싱에 대한 설명이 자세히 나와있어 참고 https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/#consumer-groups-and-rebalance-protocol

목차 [Range] Consumer를 브로커가 할당한 member_id 를 사용하여 사전순으로 배치하고, 이를 파티션 숫자 순서대로 할당. ex) consumer 3개가 있고, 2개 토픽이 각각 파티션 2개,2개를 갖고 있다면 아래처럼 할당된다. Consumer Assigned Topic Partitions C0 T0P0, T1P0 C1 T0P1, T1P1 C2 유휴 만약 토픽별로 파티션 개수의 편차가 크다면 뒤에 있는 Consumer들은 유휴상태로 남아있는다. [Round-Robin] 참조: https://kafka.apache.org/24/javadoc/org/apache/kafka/clients/consumer/RoundRobinAssignor.html#:~:text=The%20round%20rob..

목차 [목표] Auto-Commit을 할 때 방금 가져온 메시지 오프셋을 커밋하는지 마지막에 처리한 오프셋을 커밋하는지 확인. Consumer에서 Auto-Commit 을 사용할 때 메시지 유실 가능성이 있는지 확인. [메시지 유실될 가능성?] 구글링을 해보면 아래같은 케이스에서 메시지 유실 가능성이 있다고 한다. 이러한 상황이 발생하려면 auto commit을 켰을 때 방금 읽어온 메시지 오프셋을 커밋해야 한다. [테스트 환경] Java - zulu JDK11 Apache Kafka Client - 2.8.1 Spring Kafka - 2.6.13 [테스트 방법] Consumer Properties enable.auto.commit = true max.poll.records = 10 auto.commi..