[Kafka Connect] Kafka Connect Sink DB에 저장이 안 될 때

 

Kafka Connect Sink를 사용하면서 응답도 잘 오고 동작이 잘 되는 듯 싶었으나 실제로는 DB에 해당 데이터가 저장되지 않았다. 그래서 해당 현상을 해결하면서 어떤 과정을 거쳤는지 기록하고자 글을 쓰기로 했다.

 

---

 

데이터 타입 관련

 

내가 겪은 첫 번째 문제는 데이터 타입 관련 문제였다.

 

제대로 작성한 경우

 

카프카로 전달될 스키마에 포함된 필드 내용인데 처음에 객체의 타입이 Long이니까 long으로 작성하고 이런식으로 해버려서 오류가 발생했었다.

카프카 커넥트에서 지원하는 데이터 타입에 맞춰 작성해주어야 제대로 동작한다.

 

https://kafka.apache.org/24/javadoc/index.html?org/apache/kafka/connect/data/Schema.html

kafka 2.4.0 API

 

이런식으로 지원하는 데이터 타입이 따로 있다.

 

---

 

JSON 변환시 null 또는 필드명 매핑 문제

 

사실 필드명과도 관련이 있다. 자바는 일반적으로 변수명을 지을 때 camelCase를 사용한다. 그런데 DB에서는 보통 snake_case를 사용하다보니 이름이 일치하지 않아 null로 저장되는 것이다. JPA를 사용할 때는 @Column으로 필드명을 따로 지정해줄 수 있지만 카프카 커넥트에서는 따로 지원하는 어노테이션이 없는 것 같아 어쩔 수 없이 snake_case로 멤버변수명을 지정해주었다.

 

 

---

 

오류가 발생할 때 다시 세팅하기

 

이상하게 오류가 발생했을 때 나중에 정상적인 데이터로 다시 시도해도 문제가 발생했다. 그래서 토픽과 커넥터를 삭제해주는 작업이 필요했다.

 

 

 

1. 커넥터 삭제

 

 

먼저 등록해두었던 sink 커넥터를 삭제한다.

 

DELETE 메서드로 /connectors/[커넥터명]

 

으로 요청을 보내면 삭제된다.

 

 

 

 

2. 토픽 삭제

 

나는 GCP에 도커 컨테이너 내에서 카프카 커넥트를 실행하고 있기 때문에 약간의 과정을 거쳤다.

 

docker exec -it kafka bash

 

컨테이너가 실행 중인 상태에서 위 명령어로 컨테이너와 연결한다.

 

kafka-topics --delete --bootstrap-server kafka:9092 --topic [토픽명]

 

위 명령어로 토픽을 삭제한다.

 

 

 

 

 

3. 다시 커넥터 생성

 

 

http://api.delibird.store:8083/connectors

 

커넥터를 다시 생성한다.

 

 

http://api.delibird.store:8083/connectors/delibird-sink-connect/status

 

/[커넥터명]/status 로 요청을 보내서 커넥터 상태를 확인한다.

 

여기까지 진행하면 정상적으로 처리되는 것을 확인할 수 있다.

 

---

 

날짜 자동 입력 적용하기

 

카프카 커넥트로는 날짜 전달하기가 애매하기 때문에

 

 

 

DB에서 알아서 하도록 설정했다.

 

CREATE TABLE [테이블명] (
    id INT AUTO_INCREMENT PRIMARY KEY,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

 

SQL로는 이런식으로 작성하면 된다.

 

ALTER TABLE [테이블명]
ADD COLUMN created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP;

 

 

 

맨 아래가 위 과정을 전부 진행한 뒤의 결과이다.

이제 저장이 잘 된다 ㅠ.ㅠ..!