Debezium Source 에서 topic에 저장되는 UTC시간대 질문

안녕하세요.
topic에 UTC 시간대로 저장되는 문제가 있습니다.
해결 접근방법에 조언을 듣고 싶습니다.

Sink를 적용했을때 customers, products,order_items는 문제없이 적용되었으나 orders테이블의 timestamp타입의 order_datetime컬럼에 문제가 발생하여
SMT 옵션을 추가하다가 발견한 문제입니다.

결론적으로 mysql_cdc_oc_sink_orders_01.json에

        "transforms": "ConvertDateTimeType",
        "transforms.ConvertDateTimeType.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.ConvertDateTimeType.target.type": "Timestamp",
        "transforms.ConvertDateTimeType.field": "order_datetime",
        "transforms.ConvertDateTimeType.format": "yyyy-MM-dd'T'HH:mm:ss'Z'",
        "transforms.ConvertDateTimeType.timezone": "Asia/Seoul"

위 옵션을 추가하여 sink로 저장을 해결하였으나 topic에 저장되는 시간이 다르게 저장되는것을 발견했습니다.

source 데이터베이스에서는 2023-06-20 13:56:40 에 저장하였으나

sink 데이터베이스에서는 2023-06-20 04:56:40으로 저장되고 있었습니다.

이에 topic을 확인해보니 저장되는 시간대가 2023-06-20 04:56:40으로 topic에서부터 저장되는 값이 다른 것을 알 수 있었습니다.

따라서 source설정쪽이 문제일 것 같은데
"database.connectionTimeZone": "Asia/Seoul"
옵션을 넣었음에도 UTC로 적용되고있어 질문드립니다.

감사합니다.

mysql_cdc_oc_source_01.json

{
    "name": "mysql_cdc_oc_source_01",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "192.168.56.101",
        "database.port": "3306",
        "database.user": "connect_dev",
        "database.password": "connect_dev",
        "database.server.id": "10001",
        "database.server.name": "mysql01",
        "database.include.list": "oc",
        "table.include.list": "oc.customers, oc.products, oc.orders, oc.order_items",
        "database.history.kafka.bootstrap.servers": "192.168.56.101:9092",
        "database.history.kafka.topic": "schema-changes.mysql.oc",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
 
 
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false",
        "database.connectionTimeZone": "Asia/Seoul"
    }
}

MYSQL의 TIME_ZONE은 한국시간대입니다.

mysql> select @@system_time_zone;

+--------------------+

| @@system_time_zone |

+--------------------+

| KST                |

+--------------------+

transforms=convertTimezone
transforms.convertTimezone.type=io.debezium.transforms.TimezoneConverter
transforms.convertTimezone.converted.timezone=Asia/Seoul