Kafka Consumer 주요 설정

fetch.min.bytes:

• 컨슈머가 브로커에서 메시지를 가져올 때 최소 데이터 양을 지정하는 설정이다. 즉 이 값보다 더 적은 데이터는 브로커에서 가져오지 않게된다.
• 컨슈머가 브로커에서 너무 잦게 데이터를 가지고 와서 사용되는 CPU 리소스가 많다면 이 값을 올려주는 것도 방법이다.
• 이 값을 올리면 지연시간도 길어질 수 있음을 알아야한다.
• 기본값은 1Byte 이다.

 

fetch.max.wait.ms:

• 컨슈머에서 브로커로 메시지를 가지고 올 때 최대 얼마나 기다릴 것인지를 정하는 설정이다.
• 브로커에서 컨슈머로 보낼 수 있는 데이터가 fetch.min.bytes 를 만족하지 않더라도 이 시간이 지나면 컨슈머는 메시지를 가지고 온다.
• 이 값이 크면 지연시간도 증가할 수 있음을 알아야한다.
• 기본값은 500ms 이다.

 

fetch.max.bytes:

• 브로커에서 가지고 올 수 있는 최대 바이트 수를 말한다.
• 컨슈머에서 사용 가능한 메모리가 적다면 이 값을 줄이는 것이 좋다.
• 브로커에서 컨슈머로 가지고 올 때 데이터를 배치식으로 가지고 오는데 배치 안의 메시지 하나의 크기가 fetch.max.bytes 보다 크다면 이 설정에 상관 없이 해당 메시지 하나는 가지고 올 수 있다.
• 기본값은 50MB 이다.

 

max.poll.records:

• poll() 을 호출할 때마다 리턴되는 최대 레코드 수를 말한다.
• 기본값은 500 이다.

 

max.partition.fetch.bytes:

• 파티션에서 가지고 올 수 있는 최대 바이트 수를 말한다.
• 브로커에서 가지고 오는 메시지가 얼마나 많은 파티션이 담겨있는지 알 수 없기 때문에 이 값으로 메모리 제어를 하기 보다는 fetch.max.bytes 로 정하는게 낫다.
• 기본값은 1MB 이다.

 

session.timeout.ms 와 heartbeat.interval.ms:

• session.timeout.ms 기간동안 컨슈머가 브로커에게 하트비트를 보내지 않는다면 브로커는 해당 컨슈머가 죽었다고 판단하고 리밸런싱을 진행한다.
• 하트비트를 보내는 주기는 heartbeat.interval.ms 로 정할 수 있다.
• heartbeat.interval.ms 는 대체로 session.timeout.ms 값의 1/3 로 잡는 편이다.
• heartbeat.interval.ms 기본값은 3초이고, session.timeout.ms 는 45초이다. 원래는 10초였지만 이건 온프레미스 환경을 타겟한 설정이였고, 클라우드 기반의 환경이 대세이므로 이 환경에 맞게 45초로 바꿨다고 한다. 

 

max.poll.interval.ms:

• 이 시간동안 컨슈머가 브로커의 메시지를 가지고 가지 않으면 브로커는 컨슈머가 죽었다고 판단한다.
• 만약 이 설정이 없고, 컨슈머가 죽는 것을 하트비트로만 판단한다면 문제가 생길 수 있다. 하트비트는 백그라운드 스레드로 데이터를 보내기 때문에 실제 메인 스레드가 데드락이 걸려서 죽어있는 상태에서도 컨슈머는 살아있다고 판단하기 때문이다.
• 기본값은 5분이다.

 

default.api.timeout.ms:

• 컨슈머 API 를 사용하는 모든 메소드들은 이 설정 값만큼의 타임아웃을 가진다. 단 하나의 예외인 poll() 을 제외하고, poll() 은 그래서 별도로 호출할 때 타임아웃 값을 지정해줘야한다.
• 이 타임아웃은 commit, join, sync와 같은 다양한 Kafka API 호출에 적용된다.
• 만약 특정 API 호출에 대해 별도의 타임아웃 설정이 명시되지 않은 경우, 이 값이 기본적으로 사용된다.
• 기본값은 1분이다.

 

request.timeout.ms:

• 이 값은 컨슈머가 브로커에게  메시지를 가지고 오는 요청을 보낸 후 기다릴 수 있는 최대 시간이다. 이 값만큼을 기다려도 응답이 오지 않는다면 연결을 끊고 다시 재연결을 시도한다.
• 기본값은 30초인데 가능한 바꾸지 않는 것을 권장한다.
• 이 값은 브로커에 메시지를 가지고 오는 네트워크 요청에 사용되는 timeout 값을 말한다. 이렇게 말하면 poll() 에서도 인자로 타임아웃 값과 무슨 차이인지 모를 수 있는데 poll() 인자로 전달하는 타임아웃 값은 그 시간동안 기다리고 리턴되는 함수다. 데이터가 없다면 빈 값으로 리턴이 될 것이다. request.timeout.ms 는 poll() 안에서 브로커에 메시지를 가지고 올 때 타임아웃 값을 말한다. request.timeout.ms 값이 poll() 타임아웃 값보다 작다면 하나의 poll() 안에서 여러번의 브로커로 요청을 할 수 있을 것이다.

 

auto.offset.reset:

• 컨슈머가 브로커에서 메시지를 가지고 올 때 커밋된 오프셋이 없거나 (= 시간이 지나 지워진 경우), 처음 메시지를 가지고 올 때, 그리고 오프셋이 유효하지 않는 경우가 발생할 때 오프셋 리셋을 어떻게 할 지 정하는 설정이다.
• 관련값은 크게 3가지가 있다:
  • latest: 파티션에 쓰여진 메시지 중 가장 최신의 메시지를 가지고 온다.
  • earliest: 파티션에 쓰여진 메시지 중 가장 오래된 메시지를 가지고 온다.
  • none: 유효하지 않은 오프셋을 읽으려고 할 때 에러를 던진다.

 

enable.auto.commit:

• 오프셋 커밋을 자동으로 할 지, 수동으로 할 지 결정하는 옵션이다.

 

partition.assignment.strategy:

• 컨슈머에게 파티션 할당 전략을 지정하는 설정이다.
• 관련값은 다음과 같다:
  • Range:
    • 토픽 파티션을 그룹으로 나눠서 할당한다.
    • 예시로 T1, T2 토픽이 3개의 파티션이 있다면 이 파티션들의 0번, 1번, 2번 이렇게 나누고 이들을 컨슈머의 수에 맞게 그룹으로 묶어서 할당하는 전략이다.
    • 이 예시로 컨슈머가 C1, C2 이렇게 있다면 C1 은 T1, T2 의 0번과 1번 파티션을 받을 것이고, C2 는 T1, T2 의 2번 파티션을 받을 것이다.
    • 파티션이 컨슈머 수로 딱 나눠 떨어지지 않을 때, 첫번째 컨슈머가 더 많은 파티션을 할당받는 배분의 문제가 있긴한 전략이다.
  • RoundRobin:
    • 토픽 파티션들을 나열해놓고 라운드로빈으로 배분하는 방식이다.
    • 거의 똑같은 개수의 파티션을 컨슈머에게 배분한다. 
  • Sticky:
    • RoundRobin 의 파티션 배분 공정화 기능에다가 추가로 리밸런싱이 일어날 때 기존 파티션들은 가능한 같은 컨슈머에게 할당될 수 있음을 보장하는 전략이다.
  • Cooperative Sticky:
    • Sticky 방식과 동일하나 여기서는 리밸런싱 전략으로 Cooperative Rebalance 를 사용할 수 있다.
    • 가능한 이 전략을 쓰는 것이 가장 좋다.

 

client.id:

• 브로커가 컨슈머를 식별하기 위해서 쓰인다.
• 로깅, 모니터링, 쿼터 지표에서 쓰임.

 

client.rack:

• client.rack 값을 broker.rack 값 중 하나로 선택해서 가까운 브로커에 있는 파티션에서 데이터를 읽어올 수 있도록 하기 위한 설정이다.
• 원래 파티션은 리더 파티션에서만 읽어올 수 있는데, 가용 영역이 여러 곳이라면 가까운 데이터센터에서 읽도록 하는 것이 성능면에서 좋으니 이를 위한 설정이다.
• client.rack 을 설정한 이후에 브로커의 replica.selector.class 값을 RackAwareReplicaSelector 로 변경해주면 된다.
• 읽을 레플리카를 결정하는 커스텀 로직을 직접 구현하고 싶다면 ReplicaSelector 인터페이스를 구현하는 클래스를 작성하고 이를 브로커에 replica.selector.class 에 넘겨주면 된다.

 

group.instance.id:

• 컨슈머 그룹에 파티션을 정적 멤버쉽 등록을 하기 위한 설정이다.
• 정적 멤버쉽을 하게되면 잠깐 동안 컨슈머가 종료되더라도 파티션은 다른 컨슈머에게 할당되지 않게 된다.
• 물론 만료 시간도 있다. session.timeout.ms 인데 이 값을 충분히 크게 두도록 하면 정적 멤버쉽을 유지하게 할 수 있을 것이다.

 

receive.buffer.bytes, send.buffer.bytes:

• 송/수신을 위한 소켓 버퍼의 크기를 말한다.

 

offset.retention.minutes:

• 브로커 설정이지만 컨슈머 오프셋 관리에 영향을 끼치는 설정이다.
• 컨슈머 그룹에 남아있는 컨슈머가 없을 때 이 기간이 지나게 되면 컨슈머 오프셋은 삭제된다.
• 기본값은 7일이다.