[26.08.24] CDC 컨슈머 처리량을 초당 3만 건에서 8만 건으로 올리기

개요

2026년 8월 3주차 회고에 “CDC를 로컬에서 직접 테스트해보고 겪은 경험을 정리하기”를 Try로 남겼고, 그 후속으로 쓴 글이다. Debezium과 MySQL binlog로 CDC 파이프라인을 만들고 부하를 걸어 컨슈머가 못 따라가는 상황을 만들었다. 최고 lag은 115만까지 쌓였고, 거기서부터 설정과 코드를 하나씩 바꿔 소비 속도를 올렸다.

병목을 찾는 과정이 예상과 계속 달랐다. 처음에 의심한 것은 이벤트마다 찍는 INFO 로그였다. 끄니 20% 빨라지긴 했지만 그게 전부였다. 프로파일을 떠보니 CPU 시간의 54%가 JSON 파싱이었는데, 이건 설정을 바꿔서 줄일 수 있는 것이 아니었다. 필요한 필드만 읽는 파서를 60여 줄 새로 짜고 등가성 테스트를 붙이고 나서야 19.6%를 얻었다.

측정 도구도 두 번 고쳤다. 한 번은 소비 속도를 재는 구간을 잘못 잡아서, 개선한 설정이 오히려 더 느리게 측정되기까지 했다.

측정 환경은 Mac16,7(14 core / 48GB)에 Docker로 14 CPU와 8GB를 할당했고, MySQL 8.4와 Kafka 4.0(KRaft), Debezium 3.6, Spring Boot 4.1 + Kotlin을 썼다. 로컬 단일 노드에서 낸 수치이므로 운영 환경의 절대값으로 읽을 값은 아니다.

무엇을 만들었나

CDC를 공부하려고 만든 것이라 Kafka를 끼운 파이프라인의 구성만 보이면 충분했고, 도메인과 앱 구조, 저장 방식은 단순하게 뒀다. Kotlin + Spring Boot 앱 하나가 상품 가격과 재고를 REST로 변경하면, 그 변경이 MySQL binlog와 Debezium, Kafka를 거쳐 같은 앱의 @KafkaListener로 돌아와 “가격이 얼마에서 얼마로 바뀌었다”로 해석된다.

Debezium은 Embedded Engine으로 앱에 내장할 수도 있어서 Kafka 없이도 CDC 자체는 된다. 다만 그러면 알아보려던 토픽 이름 규칙과 오프셋 관리, 컨슈머 그룹, lag이 전부 사라지기 때문에 Kafka Connect 방식을 택했다.

전제 : 부하를 걸었더니 lag이 115만까지 쌓였다

처음에는 SQL 문을 파이프로 보냈는데 문장마다 네트워크 왕복이 있어 초당 1,351건이 한계였다. 이 속도로는 Debezium이 다 따라가서 lag이 전 구간 0으로 나왔다. 저장 프로시저로 루프를 MySQL 내부에서 돌리고 500건마다 커밋하자 초당 19만 건이 나왔고, 그때부터 뒤쪽이 밀리기 시작했다.

CREATE PROCEDURE gen_load(IN target_id BIGINT, IN n INT, IN span INT)
BEGIN
  DECLARE i INT DEFAULT 0;
  SET autocommit = 0;
  WHILE i < n DO
    UPDATE products SET stock_quantity = (stock_quantity + 1) % 1000000
      WHERE id = target_id + (i % span);
    SET i = i + 1;
    IF i % 500 = 0 THEN COMMIT; END IF;
  END WHILE;
  COMMIT;
END

난수를 대입하는 방식은 쓸 수 없었다. 난수가 현재 값과 우연히 같으면 그 UPDATE는 binlog 이벤트를 만들지 않는데, InnoDB가 모든 컬럼이 같은 UPDATE는 행을 실제로 쓰지 않기 때문이다. 그래서 값이 반드시 바뀌는 증가 연산으로 바꿨고, 바꾸기 전에 직접 확인해뒀다.

같은 UPDATE 문을 두 번 실행했는데 이벤트는 1건만 발행됐고, 두 번째에는 ON UPDATE CURRENT_TIMESTAMP도 발동하지 않았다. UPDATE를 실행했는데 updated_at이 그대로인 경우가 여기에 해당한다.

부하를 200만 건까지 늘려가며 측정한 결과는 아래와 같다.

lag 숫자만으로는 컨슈머가 느린 건지 생산이 늘어난 건지 모른다

lag이 늘어나는 이유는 컨슈머가 느려진 것과 생산이 폭증한 것 두 가지인데, lag 자체가 생산과 소비가 동시에 만드는 값이라 숫자만으로는 둘을 구분할 수 없다. 그래서 발행이 멈춘 시점으로 구간을 나눠 소비 속도만 따로 계산했다.

두 구간의 소비 속도가 거의 같으니 컨슈머는 내내 같은 속도로 처리한 것이고, 발행이 2배 이상 빨라서 밀린 것이다. 소진 예상 시간은 1,156,000 ÷ 30,110 ≈ 38초였고 실제로 발행이 멈춘 뒤 lag은 0까지 내려갔다. 누적 401만 건 중 유실은 없었다. 순간 유입이 소비 능력의 4배를 넘었어도 Kafka에 쌓였다가 그대로 처리됐으니, 이 lag은 장애가 아니다.

시간과 메모리, I/O를 각각 쟀다

측정은 lag이 쌓여 컨슈머가 포화된 구간에서 했다. 부하가 없으면 컨슈머가 poll()에서 대기해 프로파일이 대부분 park/wait로 차기 때문이다.

이벤트 1건 처리에 33.0µs가 소요되고 그중 절반이 대기다

jdk.ThreadCPULoad로 보니 리스너 스레드가 1코어의 약 50%만 쓰고 있었다. 절반은 CPU 실행이고 절반은 대기라는 뜻이다. JFR ExecutionSample 663개로 그 CPU 절반을 분해했다.

flame graph는 가로폭이 그 함수가 스택에 있던 샘플 수, 즉 CPU 시간 비중을 나타낸다. onMessage 위에서 readValue와 Logger.info 두 갈래로 갈라지는 것이 보인다.

이벤트당 12,635 bytes를 할당하지만 GC 때문에 느려지지는 않았다

jdk.ObjectAllocationSample의 weight를 합산하면 포화 구간에서 초당 380MB, 이벤트당 12,635 bytes였다. 상위 타입은 byte[] 31.3%, Object[] 12.6%, HeapByteBuffer 6.4%, String 6.1%로 모두 JSON 파싱 부산물이다. 다만 GC는 원인이 아니었다. 34초 구간에서 pause가 55건, 합계 52.2ms(0.15%)에 그쳤다. 할당이 많아도 대부분 단명 객체라 young GC가 짧게 끝냈다.

대기의 절반은 브로커 응답을 기다리는 85ms였다

I/O를 재는 데서 막혔다. JFR의 I/O 이벤트는 임계값 이상만 기록하는데, 리스너 스레드에서는 ThreadPark와 SocketRead가 0건이고 FileWrite가 9건(27ms, 0.1%) 뿐이었다. 긴 블로킹은 없고 짧은 대기가 무수히 많은 형태여서 프로파일로는 잡히지 않았다. spring-boot-starter-actuator를 붙여 Kafka 컨슈머 메트릭을 보고서야 원인을 찾았다.

CPU 프로파일에는 이 85ms가 잡히지 않는다.

fetch 는 브로커에서 받아오고 poll() 은 받아둔 것을 앱에 넘긴다

fetch 는 컨슈머가 브로커에 보내는 네트워크 요청이다. 응답으로 받은 레코드를 컨슈머 내부 버퍼에 채워둔다. poll() 은 애플리케이션이 호출하는 API로, 이미 버퍼에 있는 레코드를 꺼내 리스너에 넘긴다.

컨슈머가 이벤트 1건을 처리하는 순서는 이렇다.

1. KafkaConsumer.poll() 이 버퍼에서 최대 500건을 꺼낸다
2. StringDeserializer 가 byte[] 를 String 으로 바꾼다. 따로 설정하지 않아 Spring Boot 기본값이 쓰인다
3. Spring Kafka 컨테이너가 onMessage() 를 호출한다. 단건 리스너라 레코드 1건마다 한 번씩 부른다
4. record.value() 가 null 이면 tombstone 이므로 그대로 반환한다
5. objectMapper.readValue(value, DebeziumEnvelope::class.java) 가 JSON 텍스트를 객체로 만든다
6. changeDetector.detect(envelope) 가 before 와 after 를 BigDecimal.compareTo 로 비교하고 요약 문자열을 만든다
7. eventStore.add(event) 가 @Synchronized 를 붙인 ArrayDeque 에 넣는다. JavaMonitorEnter 는 0건이었다
8. log.info("[CDC] ...") 가 logback 을 거쳐 stdout 에 초당 3만 줄을 쓴다

앞의 표에서 CPU 54.1%를 쓴 JSON 파싱이 5번이고, 24.9%를 쓴 로깅이 8번이다.

여기서부터는 한 번에 하나씩만 바꿔 재고, 매번 직전 조건과 비교했다. 조건은 100만 이벤트, 파티션 4개, 부하는 상품 키 300개에 분산이다. 소비 속도는 컨슈머가 포화된 구간, 즉 첫 샘플부터 lag 최고점까지의 consumed 기울기로 계산했다.

딥다이브 1 : 이벤트당 로깅을 끄니 20% 빨라졌다

이벤트당 INFO 로깅은 CDC 이벤트를 눈으로 보려고 넣은 코드고, 실제 컨슈머라면 이벤트마다 로그를 남기지 않는다. 그런데 프로파일에서 24%를 차지할 만큼 커서, 이것을 베이스라인에 남겨두면 나머지 요인의 개선율이 부풀려진다. 코드는 그대로 두고 로그 레벨만 WARN으로 내려 log.info가 실행되지 않게 한 뒤 같은 조건에서 프로파일을 다시 떴다.

AS-IS 로깅 INFO	TO-BE 로깅 WARN

왼쪽에서 onMessage 위로 갈라지던 두 갈래 중 주황색 로깅 갈래가 오른쪽에서는 사라졌고, 남은 폭은 readValue 갈래가 차지한다. 세로 길이가 줄어든 것은 logback의 appender 체인이 통째로 없어졌기 때문이다.

JSON 파싱은 48.8%에서 72.7%로 늘었지만 파싱 자체가 느려진 것은 아니고, 로깅이 빠지면서 분모가 줄었다. 그래서 절대량으로 환산했다.

CPU는 1.61µs 줄었는데 전체 시간은 7.11µs 줄었다. 줄어든 CPU만으로 설명되지 않는 5.5µs가 로깅이 만들던 대기이고, stdout에 쓰는 동안 블로킹되는 시간이다.

이후 실험은 로깅을 끈 상태를 베이스라인으로 삼았다. 같은 조건에서 다시 재니 37,103/s였다.

딥다이브 2 : poll() 이 한 번에 꺼내는 개수를 늘리니 54% 빨라졌다

poll() 한 번이 리스너에 넘기는 레코드 수를 4배로 올렸다. 네트워크 쪽 지표는 거의 그대로였다. 응답 하나에 담겨 오는 레코드가 3,827건에서 3,761건으로, 응답 크기는 3.4MB로 유지됐다. 브로커와 주고받는 양을 하나도 바꾸지 않고 앱에 넘기는 방식만 바꿔 54%를 얻은 셈이다.

로깅을 끈 베이스라인에서 다시 재면 37,103/s에서 63,502/s로 71% 빨라진다. 레코드당 처리가 가벼워질수록 poll() 호출 자체의 비용이 차지하는 비중이 커지기 때문이다.

딥다이브 3 : 브로커에서 받아오는 양을 늘리니 4% 느려졌다

받아오는 양을 늘리면 요청 횟수가 줄어 빨라질 것이라고 봤는데 반대였다. 설정 자체는 의도대로 작동해서, 응답 하나에 담겨 오는 레코드가 3,827건에서 15,260건으로, 응답 크기가 3.4MB에서 13.5MB로 늘고 요청 횟수는 초당 2.16회에서 0.67회로 줄었다. 대신 요청부터 응답까지 걸리는 시간이 148ms에서 285ms로 늘었다.

게다가 poll() 은 여전히 500건씩만 넘기니 15,260건짜리 버퍼를 비우려면 poll() 을 30여 번 불러야 한다. 버퍼에서 꺼내 넘기는 쪽이 병목인데 받아오는 양만 키우면 버퍼만 커지고 처리량은 늘지 않는다. 조합에서도 마찬가지여서, max.poll.records만 올린 48,049/s에 이 설정을 더하니 41,414/s로 떨어졌다.

딥다이브 4 : 리스너가 레코드를 한 번에 받으니 6.5% 빨라졌다

여기부터는 설정이 아니라 코드를 바꾼 실험이다. 배치 리스너는 poll() 한 번이 가져온 레코드를 리스트로 한꺼번에 받아서, 리스너 호출을 poll() 당 1회로 줄인다. 건당 처리 내용은 같고 호출 횟수만 달라지므로 처리 방식의 효과만 분리된다. 두 리스너가 같은 코드를 쓰도록 건당 처리 로직을 CdcRecordHandler 로 떼어냈다. 둘 다 로깅을 끄고 파티션 4개, 컨슈머 스레드 1개로 측정했다.

레코드마다 반복되던 Spring Kafka 컨테이너의 호출 준비가 poll() 당 1회로 줄어든 만큼이다. 프로파일에서 Spring Kafka 계층이 6~7%였으니 그 범위 안에서 나온 값이다. 건당 파싱 비용은 그대로여서 개선폭이 작다.

딥다이브 5 : 필요한 필드만 읽으니 19.6% 빨라졌다

CPU 1순위였던 JSON 파싱을 직접 줄여봤다. 메시지를 뜯어보니 안 쓰는 부분이 절반 가까이였다.

전체 688 bytes
├─ source 311 bytes (45%)     17개 필드 중 file, pos 두 개만 쓴다
├─ before / after             실제로 필요한 부분
└─ transaction, ts_us, ts_ns  아예 안 쓴다

기존 코드는 objectMapper.readValue(json, DebeziumEnvelope::class.java) 로 문서 전체를 객체에 바인딩한다. @JsonIgnoreProperties(ignoreUnknown = true) 가 붙어 있어 안 쓰는 필드는 버리지만, 버리기 전에 토큰으로 쪼개고 필드 이름을 찾는 일은 모두 거친다. 그래서 필요한 필드만 읽고 나머지는 JsonParser.skipChildren() 으로 건너뛰는 파서를 만들어 비교했다.

둘 다 로깅을 끄고 파티션 4개, 스레드 1개, 150만 이벤트로 측정했다.

두 파서가 같은 결과를 내지 않으면 비교가 성립하지 않으므로 등가성 테스트 5건을 먼저 붙였다. UPDATE 와 DELETE, 스냅샷, 모르는 필드가 추가된 메시지에서 op 와 before, after, source, ts_ms 가 모두 같고 도메인 이벤트까지 같은 것을 확인했다.

AS-IS 객체 바인딩	TO-BE 필요한 필드만 읽기

flame graph 에서 사라진 영역

두 파서는 같은 결과를 내고, JSON 텍스트를 끝까지 읽는 것도 같다. 달라진 것은 읽어낸 값을 객체로 만드는 방법이다. 객체 바인딩에만 있고 새 파서에서는 사라진 프레임을 프로파일에서 뽑으면 이렇다.

샘플 수가 같은 짝끼리 묶으면 세 갈래이고, 각각 하는 일은 이렇다.

• _deserializeUsingPropertyBased 는 ProductRow 가 Kotlin data class 라서 생긴다. 값을 생성자에 한 번에 넘겨야 하므로 Jackson 은 PropertyValueBuffer 에 모아뒀다가 마지막에 생성자를 리플렉션으로 호출한다. 새 파서는 지역 변수에 모았다가 ProductRow(...) 를 그대로 부르니 버퍼도 리플렉션도 없다.
• SettableBeanProperty.deserialize 는 필드마다 타입에 맞는 역직렬화를 찾아 부르는 부분이다. 새 파서는 분기에서 값을 바로 읽는다.
• _deserializeWithErrorWrapping 은 실패한 필드와 이유를 알려주려고 프로퍼티마다 처리를 감싸고 경로를 쌓는다. 새 파서는 그 정보를 만들지 않는다. 처리량을 얻는 대신 진단 정보를 포기한 셈이다.

앱 코드 비중이 6.3%에서 17.6%로 늘었다. Jackson 이 하던 일을 우리 코드가 직접 하게 됐으니 늘어나는 것이 맞다. 절대량으로 환산하면 줄어든 쪽이 더 컸다.

jackson 에서 1.93µs 를 덜어내고 앱 코드로 0.21µs 를 옮겨왔으니 순수 절감이 1.72µs 다. BeanDeserializer 와 KotlinValueInstantiator 호출이 사라진 만큼 Kafka 클라이언트와 Spring 계층 측정치도 함께 내려갔다.

필드를 건너뛰어도 텍스트는 끝까지 읽어야 한다

JSON 텍스트를 앞에서부터 훑으며 여는 괄호와 필드 이름, 값 같은 단위로 끊어내는 일은 양쪽 다 그대로 한다. 프로파일에 남은 nextToken 과 _parseName, findSymbol 이 그 부분이다. 프레임 이름으로 뚜렷하게 묶이는 것만 세 갈래로 나누면 이렇다.

세 갈래의 변화를 더하면 −0.67µs 로, 파싱 전체에서 줄어든 1.93µs 보다 작다. 나머지는 이렇게 묶이지 않는 프레임에 흩어져 있다.

skipChildren() 으로 source 블록을 통째로 버리려 해도 그 블록의 닫는 괄호가 어디인지 찾아야 하고, 그러려면 중첩된 괄호를 세면서 문자를 끝까지 훑어야 한다. 그래서 파싱 비용이 3.50µs 에서 1.58µs 로 55% 줄었을 뿐 0 이 되지는 않았다. 여기서 더 줄이려면 텍스트 자체를 작게 만들어야 하고, Avro 나 Protobuf 를 쓰는 이유가 그것이다.

새 파서가 모든 프레임에서 싼 것도 아니다. ReaderBasedJsonParser.getString 은 12샘플에서 21샘플로, TextBuffer.contentsAsString 은 5샘플에서 12샘플로 늘었다. 필요한 필드마다 p.string 을 명시적으로 부르기 때문이다. 합계로는 줄었지만 부분적으로 늘어난 곳이 있다.

skipChildren 프레임 자체는 양쪽 프로파일 어디에도 잡히지 않았다. 인라인됐거나 샘플에 걸릴 만큼 오래 머물지 않았을 텐데, 단정할 근거는 없다.

여기서도 CPU 절감(2.27µs)보다 전체 시간 절감(4.42µs)이 크다. 처리가 빨라지면 버퍼를 더 빨리 비우고 다음 fetch 요청을 더 이른 시점에 보내므로 대기도 함께 줄어든다.

대신 코드가 늘어난다. 파서가 60여 줄이고, 필드가 추가되면 파서도 함께 고쳐야 한다. 등가성 테스트 없이 바꾸기는 위험하다. 객체 바인딩은 데이터 클래스에 필드 한 줄만 더하면 끝난다. 게다가 지금은 컨슈머가 발행을 따라잡은 상태라 파싱을 줄여도 lag 은 줄지 않는다. Debezium 발행 속도를 올린 뒤에 적용 여부를 다시 본다.

추가 고민할 점

dead letter topic 고려하기

파싱에 실패한 메시지는 아무 데도 남지 않고 사라진다. 컨슈머가 멈추지 않게 하려고 리스너에서 파싱 실패를 catch 하고 return 하도록 짜뒀는데, 깨진 메시지를 토픽에 직접 넣어보니 이렇게 나왔다.

WARN  failed to parse CDC message, skipping. offset=4541902 value=THIS-IS-NOT-JSON
파티션 3  current=4541902 → 4541903,  lag=0

리스너가 정상 종료했으니 Spring Kafka는 그 레코드를 처리 완료로 보고 오프셋을 커밋한다. 재시도도 없고 어디에도 남지 않는다. catch를 지워도 크게 다르지 않은데, DefaultErrorHandler가 몇 차례 재시도한 뒤 로그를 남기고 넘어가기 때문이다. dead letter topic을 붙이지 않으면 결국 버려진다.

tombstone 고려하기

상품 3번을 삭제하니 같은 key로 메시지가 2개 나왔다.

offset 6  key={"id":3}  value={"op":"d", "before":{..}, "after":null}
offset 7  key={"id":3}  value=null

앞의 것은 무엇이 삭제됐는지 before 로 알 수 있는 이벤트이고, 뒤의 것이 tombstone이다. tombstone은 로그 컴팩션 신호다. 컴팩션은 key별 최신 값만 남기는데 삭제된 레코드를 표현할 방법이 없으니, Kafka는 null value를 이 key가 삭제됐다는 표시로 쓴다. 이게 없으면 컴팩션 후에도 id=3의 옛 값이 남아 아직 있는 것으로 읽힌다.

컨슈머 입장에서 tombstone은 처리할 내용이 없어 무시하면 되는데, null 체크를 빼면 바로 다음 줄의 JSON 파싱에서 예외가 터진다.

마무리하며

설정을 모두 적용한 조합은 86,041/s로 베이스라인의 2.8배였고, 최고 lag은 3,693까지 내려갔다. 이 수치에는 컨슈머 스레드를 1개에서 4개로 올린 몫이 들어 있다. 파티션이 4개인데 스레드가 1개였으니 맞춰준 것이고, 단독으로는 147% 빨라졌다.

프로파일을 뜨는 순서가 바뀌었다. ExecutionSample은 실행 중인 스레드를 샘플링하므로 대기 시간이 빠져 있다. 그래서 ThreadCPULoad로 CPU 포화 여부를 먼저 보고, 포화가 아니면 대기 원인부터 찾은 다음에 프로파일을 떴다.

베이스라인에서 무엇을 빼는지도 결과를 바꿨다. 로깅을 남겨뒀을 때는 개선율뿐 아니라 설정끼리의 크기 비교까지 달라졌다. 여러 설정을 한꺼번에 바꿔 비교했을 때는 max.partition.fetch.bytes가 다른 설정에 가려져 효과가 있는 것처럼 보였고, 하나씩 바꿔 다시 재고 나서야 −4%인 것을 확인했다.

소비 속도를 재는 구간도 한 번 잘못 잡았다. 발행 종료 후 구간에서 쟀는데, 컨슈머가 빨라질수록 발행이 끝난 뒤 남은 이벤트가 없어서 개선한 설정이 1,390/s로 측정됐다. 컨슈머가 포화된 구간에서 재도록 바꿔 해결했다.

최고 lag 115만은 38초 만에 소진됐고 유실도 없었다. 그래서 판단할 때는 lag의 절대값보다 추세와 소진 시간(lag ÷ 소비 속도)을 봤다.

lag이 늘었을 때 쓸 순서는 이렇게 정리했다.

1. 지연 시간의 추세를 본다. 계속 벌어지는지, 발행이 멈추면 소진되는지에 따라 대응이 달라진다
2. 영향이 있는 지점부터 확인한다. 단계별 처리량을 재면 어디에서 밀리는지 나온다
3. 그 지점의 지연 시간과 프로파일에서 개선점을 찾는다. 대기가 크면 컨슈머 메트릭을, CPU가 포화면 프로파일을 본다

다음에 확인할 것은 세 가지다. 컨슈머가 발행을 따라잡으면서 병목이 Debezium 발행(약 66,500/s)으로 옮겨갔으니 max.batch.size와 프로듀서 설정을 봐야 한다. 컨슈머 스레드를 4개로 늘렸으니 CdcEventStore의 @Synchronized에 경합이 생겼는지 다시 재야 한다. 베이스라인에서는 JavaMonitorEnter가 0건이었다. dead letter topic은 학습용이라도 붙여두려 한다.