-
spring.kafka.consumer 하위 설정
1) fetch-min-bytes : Record를 fetch 해올 최소 용량. Record를 너무 잘게 가져오지 않도록 설정
2) fetch-max-wait-ms : Record를 fetch 하는 걸 기다리는 최대 시간. 너무 자주 가져오지 않도록 설정
3) request-timeout-ms : Client 단에서 Record가 들어오는 걸 기다리는 시간. 이 시간 내에 poll Request에 대해 Response가 안오면, DisconnectException 유발.
cf) max-poll-interval-ms : 이 시간 동안 poll을 수행하지 않으면, consumer가 죽었다고 판단. session-timeout-ms & heartbeat.interval.ms 의 번 외로 rebalancing 관련 설정 값 의 번 외로 rebalancing 관련 설정 값. 이외에도 rebalancing이 일어났을 때, 다시 joinGroup 요청을 보내야하는 timeout 값으로도 사용된다.
-> 내부 로직의 Infinite loop 같은 것 때문에 heartbeat는 줄 수 있지만, poll을 하지 않는 정상적인 상태가 아닌 경우를 걸러내기 위한 설정 값
-
spring.kafka.listener 하위 설정
1) poll-timeout : poll을 기다리는 시간 (ms).
2) idle-between-polls : poll 사이에 idle (쉬는) 시간 (ms).
3) idle-event-interval : Event를 일으키는 주기 (ms). Record를 fetch한 이후부터 시작
내가 사용한 방법 :
o 요구 사항 : Consumer가 Record를 지속적으로 받고 있다가, 일정 시간마다 Record를 해소하도록 해라.
-
Kafka는 Event driven Application이기 때문에, 시간에 관련된 기능은 없었다. 그래서, Timer를 따로 구현하고, Event Handler를 통해서 지속적으로 시간을 체크 했다.
-
이 때, Event Handler 관련 interval 설정이 필요했다. idle-event-interval이 Record를 fetch한 다음부터 시간을 재기 때문에, idle-betweens-polls를 적당히 주지 않으면, 데이터가 계속 들어오는 경우 중간에 Event Handler가 작동하지 못했다. (데이터가 들어오면, 다시 idle-event-interval을 0부터 세기 시작한다.)
-
fetch-max-wait-ms 와 request-timeout-ms을 너무 짧게 주면, 아래의 에러를 지속적으로 보게 된다.
[Consumer instanceId=LAUNDRY_02, clientId=T-Group, groupId=T-Group] Error sending fetch request (sessionId=2113214776, epoch=1) to node 0:
org.apache.kafka.common.errors.DisconnectException: null
-> 데이터 유실이나, Consumer 자체에 치명적인 에러는 아니고, timeout 지나도 데이터를 못가져왔다는 에러
실제 설정 값
fetch-min-bytes : 10485760
fetch-max-wait-ms : 1000
max-poll-interval-ms: 300000
request-timeout-ms : 30000
poll-timeout : 1
idle-between-polls : 1
idle-event-interval : 1
