scdf kafka 리밸런싱 이슈로 둔갑한 레디스 블로킹 이슈

개요

팀 내 DB에 추가된 데이터를 지속적으로 폴링하여 레디스에 캐싱하는 scdf 프로젝트가 있다.

설정된대로 데이터가 노출되지 않는다는 CS가 발생하여 원인을 분석하다 로그를 추적했다.

최근 여러가지 프로젝트 확장으로 인해, 추가되는 스트림 객체가 그만큼 늘었고,

그에 따라 데이터 양 대비 캐싱 처리가 느려졌을 것을 인지하고 있었다.

 

로그 속 kafka 리밸런싱 이슈

group sync scdf에서 메세지 큐로 사용하던 kafka 쪽에서 폴링이 poll interval ms를 넘기는 이슈로 인해

컨슈머 리밸런싱이 일어났고, offset 커밋이 제대로 되지 않아 과거 데이터를 계속 컨슘했다.

따라서 최신 데이터로 캐싱되는 것이 아니라 과거 데이터로 캐싱되는 이슈가 있었다.

컨슈머 그룹 리밸런싱 완료되었고 파티션을 다른 컨슈머에게 할당했기 때문에 커밋을 완료할 수 없습니다.
 이는 poll()에 대한 후속 호출 사이의 시간이 구성된 max.poll.interval.ms보다 길었음을 의미합니다.
 이는 일반적으로 폴 루프가 메시지 처리에 너무 많은 시간을 소비하고 있음을 의미합니다.
 세션 시간 초과를 늘리거나 max.poll.records를 사용하여 poll()에서 반환되는 배치의 최대 크기를 줄여 이 문제를 해결할 수 있습니다.

현재 파티션이 4개로 나누어져있으며 source로부터 데이터를 가져온 후,

데이터가 4개의 파티션으로 분배됐다.

그러나 컨슈머는 1개이며, 스레드도 1개로 운영되고 있었다.

기존에 객체가 추가되는 TPS가 높지 않았기에 1개로 설정된 거 같았다.

참고로 컨슈머와 파티션은 1:1로 매핑되거나,

컨슈머 대비 파티션이 여러개일 경우, 컨슈머 1개에 여러개가 매핑된다.

메시지 처리 속도보다 데이터 추가되는 속도가 빨랐다.

그렇기에 메시지 처리가 완료되면 다시 Poll을 요청하는데,

처리가 완료되지 않아 컨슈머 랙이 점차 늘어났고, poll 요청 시간이 max.poll.interval.ms를 넘겨버렸다.

임계치를 넘기게 되면 컨슈머 코디네이터가 컨슈머가 문제가 있음을 인지하고 컨슈머 그룹에서 제외하게 되며,

컨슈머 리밸런싱이 일어나게 된다.

그렇게 되면 기존에 컨슈머에 매핑된 파티션들이 재분배가 된다.

그렇기에 브로커는 컨슈머가 컨슘했던 데이터에 대한 offset을 커밋하지 못하는 것이다.

물론 현재 설정된 컨슈머는 1개니까 max.poll.interval.ms 을 매우 큰 수치로 늘리면 되지 않느냐라고 할 수도 있다.

그러나 max.poll.interval.ms를 늘리는 것은

메시지 처리 완료 시간의 threshold를 늘린 것이고, 결국 추가된 데이터에 대한 캐싱 속도를 늦추게 되어

반실시간이 아닌 분 단위로 실행되는 배치와 다름이 없어지므로 근본적인 해결은 아니다.

이렇게 되면 파티션을 처리하는 컨슈머를 늘리거나 처리량을 높여야 했다.

코드 속 Redis 블로킹 이슈

사진 출처:  https://velog.io/@jsb100800/redis-pipelining

사진 출처: https://velog.io/@jsb100800/redis-pipelining

기존 코드를 보니, redis 쪽에 데이터를 넣는 작업에서 하나하나 연결을 다시 맺고 저장 요청하는 방식이었으며,

이전 작업이 처리되어야 다음 작업이 처리되는 블로킹으로 인한 이슈였다.

과거엔 스트림 데이터 양이 적었기 때문에 블로킹이 적어 감당할 수 있었으나,

데이터가 늘어나면서 다음 요청이 지연되었다.

컨슈머 랙이 점차 벌어지더니, 데이터 캐싱 작업도 늦어지고 max.poll.interval.ms를 넘길 정도가 되면서

리밸런싱이 발생되어 offset커밋이 안되는 것이었다.

해결

따라서 executePipelined 명령어를 사용하여,

연결 상태에서 한꺼번에 저장 요청하는 방식으로 수정하여 처리 속도를 높일 수 있었다.

사실 기존에는 잘 작동하고, 문제 없던 코드라서 코드를 제대로 확인하지 않고, 컨슈머 스레드를 늘려야하나 했었다.

그러나 근본적인 원인은 레디스에 커넥션을 다시 맺어 네트워크 병목현상으로 인한 이슈였던 것이다.

문제 없던 코드더라도 데이터가 늘어남에 따라 이슈가 발생하므로 섬세하게 살펴봐야한다는 것과

executePipelined를 사용해야하는 중요성을 새삼 깨달았다.