안녕하세요:) 오늘은 Debezium을 사용하여 프로젝트를 진행했을 때 발생했던 문제점에 대해 나누고자 합니다.
어떤 문제점이 있었을까요?
백문이 불여일견이므로 프로젝트를 동작하면서 문제점을 파악해보겠습니다
CQRS에 대한 생각 - 5에서 Sink-Connector설정을 보면
"auto.create": "true"
을 true로 설정했었는데 만약 true로 설정한 경우 어떻게 되는지 보도록 하겠습니다.
Sink DB
mysql> use harmony;
Database changed
mysql> show tables;
Empty set (0.01 sec)Source DB
mysql> use harmony;
Database changed
mysql> show tables;
+-------------------+
| Tables_in_harmony |
+-------------------+
| board |
| category |
| category_read |
| channel |
| channel_read |
| comment |
| emoji |
| emoji_user |
| guild |
| guild_read |
| guild_user |
| image |
| room |
| room_user |
| user |
| user_read |
+-------------------+
16 rows in set (0.02 sec)Sink DB는 Table도 생성되지 않은 상태입니다.
이때, API 호출을 통해 Source DB에 Data를 INSERT를 해보겠습니다.
현재 보이는 이미지는 길드를 생성하는 API 테스트 창입니다.
길드를 생성하는 API에는 5개의 Table에 Data를 Insert하게 되는데요. 한번 동작해보겠습니다.
잘 동작이 되었다는 것을 볼 수 있습니다.
이번에는 토픽 목록을 보도록 하겠습니다.
토픽 목록 조회
kafka-topics.sh --list --bootstrap-server kafka:9092
테이블마다 토픽이 생성된 것을 볼 수 있습니다
이번에는 Sink-Connector의 상태를 체크해보도록 하겠습니다
Sink-Connector 상태 확인
curl -X GET http://localhost:8083/connectors/sink-connector/status
잘 동작하고 있는 것을 볼 수 있습니다
그리고 Sink DB Table을 봐도 데이터 삽입이 잘 동작했다는 것을 알 수 잇습니다.
mysql> show tables;
+-------------------+
| Tables_in_harmony |
+-------------------+
| channel |
| channel_read |
| guild |
| guild_read |
| guild_user |
| user |
| user_read |
+-------------------+
7 rows in set (0.01 sec)
mysql> select * from guild;
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
| guild_id | manager_id | created_at | invite_code | name | profile |
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
| 1 | 1 | 2024-04-02 13:17:20 | 1597e41bbd9c4e2e9dc24991f4d9a8f4 | example | https://storage.googleapis.com/remember-harmony/8de81dc4-0fc8-4894-ad84-a890ba0a876e |
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
mysql> select * from guild_user;
+----------+---------------+---------+
| guild_id | guild_user_id | user_id |
+----------+---------------+---------+
| 1 | 1 | 1 |
+----------+---------------+---------+
1 row in set (0.01 sec)하지만 제가 처음에 설계했던 ERD처럼 테이블이 잘 설계되었을까요?
먼저 인덱싱을 설정했는데 잘 되어있는지 확인해보겠습니다.
Source DB
mysql> show index from guild_read;
+------------+------------+-------------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| Table | Non_unique | Key_name | Seq_in_index | Column_name | Collation | Cardinality | Sub_part | Packed | Null | Index_type | Comment | Index_comment | Visible | Expression |
+------------+------------+-------------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| guild_read | 0 | PRIMARY | 1 | guild_read_id | A | 1 | NULL | NULL | | BTREE | | | YES | NULL |
| guild_read | 1 | idx__userId | 1 | user_id | A | 1 | NULL | NULL | YES | BTREE | | | YES | NULL |
+------------+------------+-------------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
2 rows in set (0.03 sec)Sink DB
mysql> show indexes from guild_read;
+------------+------------+----------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| Table | Non_unique | Key_name | Seq_in_index | Column_name | Collation | Cardinality | Sub_part | Packed | Null | Index_type | Comment | Index_comment | Visible | Expression |
+------------+------------+----------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| guild_read | 0 | PRIMARY | 1 | guild_read_id | A | 1 | NULL | NULL | | BTREE | | | YES | NULL |
+------------+------------+----------+--------------+---------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
1 row in set (0.01 sec)Source DB를 따라서 인덱싱이 설정되지 않았다는 것을 볼 수 있습니다.
이번에는 FK 설정이 잘 되었는지 확인해보겠습니다.
Source DB
mysql> select * from information_schema.table_constraints where table_name = 'guild_user';
+--------------------+-------------------+-----------------------------+--------------+------------+-----------------+----------+
| CONSTRAINT_CATALOG | CONSTRAINT_SCHEMA | CONSTRAINT_NAME | TABLE_SCHEMA | TABLE_NAME | CONSTRAINT_TYPE | ENFORCED |
+--------------------+-------------------+-----------------------------+--------------+------------+-----------------+----------+
| def | harmony | PRIMARY | harmony | guild_user | PRIMARY KEY | YES |
| def | harmony | FKb9t28734xyf955pp93ln7kukc | harmony | guild_user | FOREIGN KEY | YES |
| def | harmony | FKndxotjttcabs12j8849yb4ydy | harmony | guild_user | FOREIGN KEY | YES |
+--------------------+-------------------+-----------------------------+--------------+------------+-----------------+----------+
3 rows in set (0.00 sec)Sink DB
mysql> select * from information_schema.table_constraints where table_name = 'guild_user';
+--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
| CONSTRAINT_CATALOG | CONSTRAINT_SCHEMA | CONSTRAINT_NAME | TABLE_SCHEMA | TABLE_NAME | CONSTRAINT_TYPE | ENFORCED |
+--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
| def | harmony | PRIMARY | harmony | guild_user | PRIMARY KEY | YES |
+--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
1 row in set (0.03 sec)FK 설정도 제대로 되지 않았다는 것을 볼 수 있습니다
왜 Sink DB에는 적용되지 않았을까요?
Sink DB에 Data가 INSERT되는 것은 카프카 토픽에 저장된 Data 이기 때문입니다.
root@a73991efa36c:/opt/kafka_2.13-2.8.1# kafka-console-consumer.sh --topic dbserver1.harmony.guild_user --bootstrap-server localhost:9092 --from-beginning
{"schema":
{"type":"struct","fields":[{"type":"int64","optional":true,"field":"guild_id"},
{"type":"int64","optional":false,"field":"guild_user_id"},
{"type":"int64","optional":true,"field":"user_id"}],
"optional":false,"name":"dbserver1.harmony.guild_user.Value"},
"payload":{"guild_id":1,"guild_user_id":1,"user_id":1}}Kafka Topic 내용을 보면 각 스키마의 데이터 타입과 구조, 필드, 값만 정의되어 있고 FK인지, Index를 인지 정의하지 않습니다.
그래서, 이러한 데이터 삽입만 이뤄질 뿐 Index, FK 같은 정보를 삽입되지 않습니다.
이번에는 Sink DB에도 Schema를 삽입 후 데이터 삽입을 해보겠습니다
Source DB
mysql> show tables;
+-------------------+
| Tables_in_harmony |
+-------------------+
| board |
| category |
| category_read |
| channel |
| channel_read |
| comment |
| emoji |
| emoji_user |
| guild |
| guild_read |
| guild_user |
| image |
| room |
| room_user |
| user |
| user_read |
+-------------------+
16 rows in set (0.03 sec)Sink DB
mysql> show tables;
+-------------------+
| Tables_in_harmony |
+-------------------+
| board |
| category |
| category_read |
| channel |
| channel_read |
| comment |
| emoji |
| emoji_user |
| guild |
| guild_read |
| guild_user |
| image |
| room |
| room_user |
| user |
| user_read |
+-------------------+
16 rows in set (0.02 sec)이번에도 똑같이 API 호출을 해보겠습니다
root@a73991efa36c:/opt/kafka_2.13-2.8.1# curl -X GET http://localhost:8083/connectors/sink-connector/status
{"name":"sink-connector","connector":{"state":"RUNNING","worker_id":"172.19.0.4:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.19.0.4:8083"}],"type":"sink"}Sink Connector의 상태도 정상이고 Data도 잘 삽입되었다는 것을 볼 수 있습니다.
mysql> select * from guild_user;
+----------+---------------+---------+
| guild_id | guild_user_id | user_id |
+----------+---------------+---------+
| 1 | 1 | 1 |
+----------+---------------+---------+
1 row in set (0.00 sec)
mysql> select * from guild;
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
| guild_id | manager_id | created_at | invite_code | name | profile |
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
| 1 | 1 | 2024-04-02 14:10:54 | 037adaf0ba1c434e9f76268aa85ea1e9 | example | https://storage.googleapis.com/remember-harmony/4cbae073-b3b8-467b-a2a4-3fdade04dad3 |
+----------+------------+---------------------+----------------------------------+---------+--------------------------------------------------------------------------------------+
1 row in set (0.02 sec)하지만, 연속으로 호출하면 어떻게 될까요?
import http from "k6/http";
import {sleep} from "k6";
import {FormData} from 'https://jslib.k6.io/formdata/0.0.2/index.js';
const img = open('/Users/0chord/Desktop/discord.png', 'b');
export const options = {
stages: [
{duration: '10s', target: 10},
{duration: '20s', target: 20},
{duration: '15s', target: 15},
],
thresholds: {http_req_duration: ['avg<100', 'p(95)<200']},
noConnectionReuse: true,
userAgent: 'MyK6UserAgentString/1.0',
};
export default function () {
const url = 'http://localhost:8000/api/community/register/guild';
// JSON 데이터 생성
const json = JSON.stringify({managerId: 1, name: "example"});
// 멀티파트 폼 데이터 생성
const formData = new FormData();
formData.append('registerGuildRequest', http.file(json, 'registerGuildRequest.json', 'application/json'));
formData.append('profile', http.file(img, 'image1.png', 'image/png'));
const params = {
headers: {
'Content-Type': 'multipart/form-data; boundary=' + formData.boundary,
},
};
// 요청 전송
const res = http.post(url, formData.body(), params);
// response 확인을 위해 sleep 사용
sleep(1);
}K6 부하테스트를 통해 연속으로 길드를 등록하는 상황을 보도록 하겠습니다
k6 run register-guild.js결과
[2024-04-02 05:20:55,947] WARN Write of 6 records failed, remainingRetries=10 (io.confluent.connect.jdbc.sink.JdbcSinkTask:101)
java.sql.BatchUpdateException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:223)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchSerially(ClientPreparedStatement.java:813)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:416)
at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:802)
at io.confluent.connect.jdbc.sink.BufferedRecords.executeUpdates(BufferedRecords.java:196)
at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:186)
at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:80)
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:90)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:582)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:330)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:237)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.sql.SQLIntegrityConstraintViolationException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:118)
at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:912)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1054)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchSerially(ClientPreparedStatement.java:792)
... 17 more
[2024-04-02 05:20:55,949] INFO Initializing JDBC writer (io.confluent.connect.jdbc.sink.JdbcSinkTask:65)
[2024-04-02 05:20:55,949] INFO Validating JDBC URL. (io.confluent.connect.jdbc.dialect.DatabaseDialects:171)
[2024-04-02 05:20:55,949] INFO Validated JDBC URL. (io.confluent.connect.jdbc.dialect.DatabaseDialects:174)
[2024-04-02 05:20:55,950] INFO Validating JDBC URL. (io.confluent.connect.jdbc.dialect.DatabaseDialects:171)
[2024-04-02 05:20:55,950] INFO Validated JDBC URL. (io.confluent.connect.jdbc.dialect.DatabaseDialects:174)
[2024-04-02 05:20:55,950] INFO Initializing writer using SQL dialect: MySqlDatabaseDialect (io.confluent.connect.jdbc.sink.JdbcSinkTask:72)
[2024-04-02 05:20:55,950] INFO JDBC writer initialized (io.confluent.connect.jdbc.sink.JdbcSinkTask:74)
[2024-04-02 05:20:55,950] ERROR WorkerSinkTask{id=sink-connector-0} RetriableException from SinkTask: (org.apache.kafka.connect.runtime.WorkerSinkTask:601)
org.apache.kafka.connect.errors.RetriableException: java.sql.SQLException: Exception chain:
java.sql.BatchUpdateException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
java.sql.SQLIntegrityConstraintViolationException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:118)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:582)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:330)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:237)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.sql.SQLException: Exception chain:
java.sql.BatchUpdateException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
java.sql.SQLIntegrityConstraintViolationException: Cannot add or update a child row: a foreign key constraint fails (`harmony`.`channel`, CONSTRAINT `FKmj3yu54l6fa5lr3pv7sh1d0dn` FOREIGN KEY (`guild_id`) REFERENCES `guild` (`guild_id`))
at io.confluent.connect.jdbc.sink.JdbcSinkTask.getAllMessagesException(JdbcSinkTask.java:165)
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:111)
... 11 more에러가 발생했습니다! 왜 발생했을까요?
바로 FK 문제 때문입니다.
갑자기 가만히 있던 FK가 문제라니... 이해가 안됩니다
하지만 유추해 보면 Kafka Connect방식 때문이라고 생각하는데요. 한번 알아보겠습니다
우리는 테이블별로 topic이 만들어진다는 것을 알고 있습니다.
root@a73991efa36c:/opt/kafka_2.13-2.8.1# kafka-topics.sh --list --bootstrap-server kafka:9092
__consumer_offsets
connect-configs
connect-offsets
connect-status
dbhistory.harmony
dbserver1
dbserver1.harmony.channel
dbserver1.harmony.channel_read
dbserver1.harmony.guild
dbserver1.harmony.guild_read
dbserver1.harmony.guild_user
dbserver1.harmony.user
dbserver1.harmony.user_read우리가 Source DB에 INSERT한 Data는 각 토픽에 적재되고 Sink DB에 저장됩니다
그런데, 각 토픽은 순서보장이 될까요?
각 파티션 내에서의 순서는 보장되지만, 파티션 간에는 순서가 보장되지 않습니다.
하지만, 저는 파티션이 1개인 경우를 가정한 후 프로젝트를 진행했으므로 순서가 보장됩니다.
토픽끼리의 순서보장이 될까요? 토픽끼리의 순서보장은 되지 않습니다. 그래서 문제가 발생했습니다.
예를 들어, 제가 설계했던 ERD에 따르면 USER, GUILD Table이 부모 Table이고 GUILD_USER는 자식 Table입니다.
그리고, Guild Table이 부모 Table이고 Channel Table은 자식 Table입니다.
그러므로 Guild, User Table에 먼저 Data가 저장된 후 Guild_User Table과 Channel Table이 저장되어야 합니다.
그런데, Topic간에는 순서보장이 되지 않으므로,
Channel과 Guild_User가 먼저 저장될 수도 있습니다.
그래서, 위와 같은 오류가 발생하면서 Sink-Connector가 동작하지 않았습니다.
해결 방안
저는 외래키 제약조건을 비활성화를 하는 방향으로 진행했습니다.
데이터 무결성, 일관성의 문제등이 발생할 수 있지만 전체적인 트랜잭션이 보장되는 환경에서 테스트 후 외래키 제약조건을 비활성화하기 때문에 문제가 발생하지 않을 것 같아 진행했습니다.
외래키 제약조건을 비활성화한 후 K6 부하테스트를 작동해보겠습니다
k6 run register-guild.js(base) ✘ 0chord@0Chord ~/Desktop/recamp/k6-script k6 run register-guild.js
/\ |‾‾| /‾‾/ /‾‾/
/\ / \ | |/ / / /
/ \/ \ | ( / ‾‾\
/ \ | |\ \ | (‾) |
/ __________ \ |__| \__\ \_____/ .io
execution: local
script: register-guild.js
output: -
scenarios: (100.00%) 1 scenario, 20 max VUs, 1m15s max duration (incl. graceful stop):
* default: Up to 20 looping VUs for 45s over 3 stages (gracefulRampDown: 30s, gracefulStop: 30s)
data_received..................: 203 kB 4.4 kB/s
data_sent......................: 22 MB 470 kB/s
http_req_blocked...............: avg=564.85µs min=125µs med=352µs max=6.38ms p(90)=908µs p(95)=1.96ms
http_req_connecting............: avg=499.02µs min=104µs med=302µs max=6.33ms p(90)=778µs p(95)=1.74ms
✗ http_req_duration..............: avg=319.11ms min=234.89ms med=272.1ms max=1.83s p(90)=606.08ms p(95)=676.86ms
{ expected_response:true }...: avg=319.11ms min=234.89ms med=272.1ms max=1.83s p(90)=606.08ms p(95)=676.86ms
http_req_failed................: 0.00% ✓ 0 ✗ 457
http_req_receiving.............: avg=699.55µs min=21µs med=528µs max=6.29ms p(90)=1.07ms p(95)=1.53ms
http_req_sending...............: avg=126.18µs min=30µs med=76µs max=5.53ms p(90)=165.2µs p(95)=240.59µs
http_req_tls_handshaking.......: avg=0s min=0s med=0s max=0s p(90)=0s p(95)=0s
http_req_waiting...............: avg=318.28ms min=234.42ms med=271.42ms max=1.82s p(90)=605.45ms p(95)=676.46ms
http_reqs......................: 457 9.926697/s
iteration_duration.............: avg=1.36s min=1.27s med=1.31s max=2.87s p(90)=1.65s p(95)=1.71s
iterations.....................: 457 9.926697/s
vus............................: 4 min=1 max=20
vus_max........................: 20 min=20 max=20
성공적으로 테스트가 실행된 것을 볼 수 있고 Sink-Connector의 상태를 보도록 하겠습니다
root@a73991efa36c:/opt/kafka_2.13-2.8.1# curl -X GET http://localhost:8083/connectors/sink-connector/status
{"name":"sink-connector","connector":{"state":"RUNNING","worker_id":"172.19.0.4:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.19.0.4:8083"}],"type":"sink"}커넥터도 잘 작동하는 것을 볼 수 있습니다
지금까지 웹 서버에서 DB접근시 Kafka Connect의 문제점과 해결방안에 대해 알아보았습니다.
다음에는 Redis를 읽기 캐시 DB로 두어 읽기 속도 향상 방법에 대해 알아보고자 합니다.
하지만, 아직 어떻게 해야할지 갈피를 잡지 못하고 있어서 좀 오래 걸릴 것 같습니다!
다음에 뵈어요 뿅!
백엔드 개발자 지망생입니다!