Kafka Connect로 파이프라인을 구성하다 보면 에러 처리를 DLQ(Dead Letter Queue)로 해결하려는 경우가 많습니다. 그런데 소스 커넥터에도 DLQ가 똑같이 적용될 거라고 생각하면 착오가 생깁니다.
DLQ는 싱크 파이프라인 전용이다
DeadLetterQueueReporter는 싱크 커넥터에서만 동작합니다. 컨슈머가 카프카에서 레코드를 읽어 컨버터, 트랜스폼을 거쳐 태스크로 전달되는 과정에서 문제가 생긴 레코드를 별도 토픽으로 보내는 구조입니다. 소스 커넥터는 반대 방향, 즉 외부 시스템에서 카프카로 쓰는 쪽이라 이 메커니즘이 적용되지 않고, 소스 커넥터의 오류는 기본적으로 태스크 실패로 처리됩니다.
DLQ로 보내진 레코드에는 __connect.errors. 접두사가 붙은 헤더로 원본 정보가 함께 남습니다. 토픽, 파티션, 오프셋, 커넥터 이름, 태스크 ID, 어느 처리 단계(Stage)에서 실패했는지, 예외 클래스와 메시지, 스택 트레이스까지 포함되어 있어서 장애 분석에 바로 활용할 수 있습니다.
errors.tolerance가 none이면 DLQ 토픽을 설정해도 무의미하다
ConnectorConfig.java를 보면 errors.tolerance의 기본값은 none입니다. 이 상태에서는 DLQ 토픽을 지정해도 아무 효과가 없습니다. none에서는 어떤 에러든 즉시 태스크 실패로 이어지기 때문에, DLQ가 실제로 동작하려면 errors.tolerance=all을 반드시 함께 켜야 합니다.
errors.tolerance=all
errors.deadletterqueue.topic.name=my-connector-dlq
errors.log.enable=true
errors.retry.timeout=60000
errors.retry.delay.max.ms=10000
errors.tolerance=all을 켜는 순간부터는 문제가 되는 레코드를 건너뛰는 방식으로 동작이 바뀝니다. 여기서 주의할 점은, DLQ 토픽 없이 톨러런스만 all로 켜두면 문제 레코드가 소리 없이 스킵되어버립니다. errors.log.enable=true까지 같이 켜서 최소한 로그로는 흔적이 남게 하고, DLQ 토픽은 별도로 소비하는 운영 루틴을 만들어야 방치된 채로 데이터가 새는 걸 막을 수 있습니다.
SinkTask에서 배치를 버퍼링한다면 flush()가 필수다
SinkTask는 initialize → start → open → put → flush → close → stop 생명주기를 가집니다. put()에서 레코드를 즉시 쓰지 않고 배치로 모아뒀다가 나중에 쓰는 구현이라면, 반드시 flush()나 preCommit()을 오버라이드해서 실제로 다운스트림에 반영된 레코드까지만 오프셋 커밋이 이뤄지도록 해야 합니다.
이걸 빠뜨리면, 아직 다운스트림에 쓰이지 않은 레코드의 오프셋이 먼저 커밋되어버리는 경우가 생깁니다. 이 상태에서 태스크가 죽으면 해당 레코드는 이미 커밋된 것으로 취급되어 재처리 대상에서 빠지고, 결과적으로 데이터 유실로 이어집니다. Javadoc에서 이 부분을 명시적으로 경고하는 이유입니다.
정리
싱크 커넥터를 새로 붙일 때는 errors.tolerance=all + DLQ 토픽 + errors.log.enable=true를 기본 조합으로 가져가는 편이 안전합니다. 소스 커넥터는 애초에 DLQ 대상이 아니라는 것을 전제로 오류 처리 전략을 따로 세워야 하고, 배치 버퍼링을 하는 싱크 태스크는 flush()/preCommit() 구현 여부를 코드 리뷰 시점에 반드시 확인하는 게 좋습니다. 분산 모드에서 커넥터/오프셋/상태를 저장하는 내부 토픽은 사용자가 직접 손대지 않는 것도 함께 챙길 부분입니다 — 잘못된 파티션 수나 압축 정책으로 재생성하면 클러스터 전체 커넥터 상태가 깨질 수 있습니다.