[CDC] Trouble Shooting, 느낀점

masibasi·2023년 5월 16일

TROUBLESHOOTING

Docker, Kafka, Debezium Engine으로 CDC 구현하기

목록 보기

6/6

TroubleShooting

가장 많이 발생한 에러 :

Docker-Compose 세부 설정 관련
REST API로 Connector 생성시 config 세부 설정 관련
KAFKA 구독중인 topic 관련
SPRING - REACT 통신 관련

처음 맞딱뜨려 당황했던 에러, 고치는데 힘들었던 에러 위주로 작성하였습니다.

#1 Docker-compose m1칩 에러

**[WARNING: The requested image’s platform (linux/amd64) does not match the detected host platform (linux/arm64/v8)](https://collabnix.com/warning-the-requested-images-platform-linux-amd64-does-not-match-the-detected-host-platform-linux-arm64-v8/)**

도커관련 이해도가 낮아서 발생했던 문제
Image가 맥에 호환되지 않는 문제. Docker-Compse 파일에서 platform: linux/amd64/v8 로 따로 명시해주어 해결
https://collabnix.com/warning-the-requested-images-platform-linux-amd64-does-not-match-the-detected-host-platform-linux-arm64-v8/

#2 Docker-compose Volume error

Error response from daemon: invalid volume specification: 'D:/mysql-sink/data:/var/lib/mysql:rw’

Volume에 없는 경로를 지정해서 발생한 에러. Docker-Compose 파일을 전혀 이해하지 않고 복사 붙여넣기 하다가 발생한 에러이다.
: 전 경로가 실제 내 디렉토리에 위치하도록 다시 확인해주어야 한다.

#3 Linux vim

리눅스에 익숙하지 않아 발생한 문제

bash: vim: command not found

apt update
apt install vim

#4 JDBC connector 설치시 No suitable driver found

https://wecandev.tistory.com/111

커넥터 설치를 위한 추가파일을 갖고있지 않아서 발생한 문제
Connect/J JDBC driver for MySQL를 다운받는다.
해당 jar 파일을 Confluent의 connect 플러그인이 설치된 디렉토리에 넣는다.
파일명 : mysql-connector-java-8.0.27.jar
경로 : {KAFKA_HOME}/connectors/confluentinc-kafka-connect-jdbc-10.2.5/lib/

docker cp mysql-connector-java-8.0.27.jar kafka:/opt/kafka_2.13-2.8.1/connectors/confluentinc-kafka-connect-jdbc-10.7.0/lib/

#5 DELETE의 경우 CDC 반영이 안된다

Sink Connector, Source Connector에 각각 config를 새로 추가해준다.

curl --location --request DELETE 'http://localhost:8083/connectors/source-test-connector'
curl --location --request DELETE 'http://localhost:8083/connectors/sink-test-connector'

curl --location --request POST 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "source-test-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "mysqluser",
    "database.password": "mysqlpw",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.allowPublicKeyRetrieval": "true",
    "database.include.list": "testdb",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "dbhistory.testdb",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",
    "transforms": "unwrap,addTopicPrefix",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.addTopicPrefix.regex":"(.*)",
   ** "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",**
    "transforms.addTopicPrefix.replacement":"$1"
  }
}'

# Sink
curl --location --request POST 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "sink-test-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://mysql-sink:3306/sinkdb?user=mysqluser&password=mysqlpw",
    "auto.create": "false",
    "auto.evolve": "true",
    "delete.enabled": "true",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "table.name.format":"${topic}",
    "tombstones.on.delete": "true",
    "connection.user": "mysqluser",
    "connection.password": "mysqlpw",
    "topics": "dbserver1.testdb.accounts",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",
    "transforms": "unwrap, route, TimestampConverter",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    **"transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",**
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "$3",
    "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
    "transforms.TimestampConverter.format": "yyyy-MM-dd HH:mm:ss",
    "transforms.TimestampConverter.target.type": "Timestamp",
    "transforms.TimestampConverter.field": "update_date"
  }
}'

#5-1 __deleted column 생성으로 인한 TOPIC 오류

위 #5의 솔루션대로 따라가면 sinkDB에는 새로운 __deleted 컬럼이 생긴다.

TOPIC에는 정상적으로 저장이 되나, SINKDB에서 이를 받아오지 못하는 오류

새로 생긴 컬럼을 생성하지 못하게 막아놓았기 떄문.

SINK CONNECTOR config를 수정해서 다시 생성해주면 정상적으로 업데이트가 된다.

curl --location --request POST 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "sink-test2-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://mysql-sink2:3306/sinkdb2?user=mysqluser&password=mysqlpw",
    "auto.create": "false",
    **"auto.evolve": "true", // 새로운 column이 생기면 이를 sinkdb에도 동일하게 생성해준다.**
    "delete.enabled": "true",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "table.name.format":"${topic}",
    "tombstones.on.delete": "true",
    "connection.user": "mysqluser",
    "connection.password": "mysqlpw",
    "topics": "dbserver1.testdb.CUSTOM_TABLE",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",
    "transforms": "unwrap, route, TimestampConverter",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "true",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "$3",
    "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
    "transforms.TimestampConverter.format": "yyyy-MM-dd HH:mm:ss",
    "transforms.TimestampConverter.target.type": "Timestamp",
    "transforms.TimestampConverter.field": "update_date"
  }
}'

#6 (SINK) 새로운 TABLE 생성시 전체TABLE 업데이트 안됨

두가지 테이블을 토픽으로 관리하려다 생긴 에러

Source db에서 새로운 테이블 DB_TEST를 생성했다. 이를 까먹은채로 accounts 테이블에 insert했을때 카프카 topic으로 생성은 되지만 sink DB에는 연동이 안되던 문제

DB_TEST는 무언가 만족하지 않는 조건이 있는 것 같았다.

SINK DB에서 모든 토픽을 구독하기로 되어있었기 때문이다.

필요한 토픽만 받아올 수 있도록 Config를 수정하였다

Debezium Connector Config를 보면

"database.include.list": "testdb",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "dbhistory.testdb",

데이터베이스 전체를 보내게 되어있다.

Sink Connector Config를 보면

    "topics.regex": "dbserver1.testdb.(.*)",
//    "topics": "dbserver1.testdb.accounts", <- 하나만 명시
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",

위처럼 전부 받게 되어있다. 일단 Sink DB에는 accounts table만 받으면 되어서 config에 하나만 받도록 수정했더니 정상적으로 accounts table을 받게 되었다.

8. (REACT, SPRING) Axios 사용시 CORS 에러

Cross Origin Resource Sharing의 약자로,

현재 웹페이지 도메인에서 다른 웹페이지 도메인으로 리소스가 요청되는 경우를 말한다.

예를 들면, 웹페이지인 http://web.com에서 API 서버 URL인 http://api.com 도메인으로 API를 요청하면

http 형태로 요청이 되므로 브라우저 자체에서 보안 상 이유로 CORS를 제한하게 되는 현상을 말합니다.

Client측에서 해결해보려한 시도 (실패) https://woochan-dev.tistory.com/94 https://www.datoybi.com/http-proxy-middleware/
Server측에서 해결해보려한 시도 (성공) WebConfig 파일을 만들어줘서 로컬 주소로부터 오는 API호출을 허용해줬다.