Flink-CDC 在同步字段为timestamp类型的数据时，初始化数据和增量数据时区不一致

在使用flinkcdc stream处理时，字段类型为timestamp的字段，初始化的数据和增量binlog的数据里的timestamp类型数据时区不一致。我是临时通过isSnapshot这个字段处理了一下，希望可以修复一下这个bug。flink版本1.2.0，flinkcdc1.4.0

[leonardBang]

@xiaoshijiang523 Could you post a simple demo to reproduce the issue? thanks

[wuqingquan-github]

env.addSource(getCDC(DATABASE, TABLE)).print();

mysql中数据：start_time和end_time都是datetime类型
{"id_card":"362324","end_time":2021-08-14 08:55:11,"start_time":2021-08-14 08:44:08,"id":72}

不管是否添加配置：SERVER_TIME_ZONE=Asia/Shanghai，结果都一样
"data":{"id_card":"362324","end_time":1628931311000,"start_time":1628930648000,"id":72}
打印的结果转换为LocalDateTime分别是2021-08-14 16:44:08和2021-08-14 16:55:11

public static DebeziumSourceFunction<JSONObject> getCDC(String databaseList, String tableList) {
    return MySQLSource.<JSONObject>builder()
            .hostname(HOSTNAME)
            .port(3306)
            .username(USERNAME)
            .password(PASSWORD)
            .databaseList(databaseList)
            .tableList(tableList)
            .serverTimeZone(SERVER_TIME_ZONE)
            .deserializer(new MyDeserializer())
            .build();
}

[ZhuJiang121]

我遇到了类似的问题，我的 mysql 本地库的时区是 UTC +8 时区，我的 datetime 类型与 timestamp 类型字段都比原字段时间多了 8 小时。我的版本： flink 1.13.1， flink cdc 2.0.0

CREATE TABLE source_user_info ( UserId INT, UserName STRING, IsAdmin TINYINT, LastLogin TIMESTAMP(3), TestTime TIMESTAMP(3),PRIMARY KEY( UserId ) NOT ENFORCED ) WITH (
'connector' = 'mysql-cdc',
'scan.startup.mode' = 'initial',
'hostname' = '172.16.10.138',
'port' = '3306',
'username' = 'root',
'password' = 'root',
'database-name' = 'hcdsj',
'table-name' = 'user_info',
'server-time-zone' = 'Asia/Shanghai',
'debezium.snapshot.locking.mode' = 'none'

当我使用 flink 1.12.2 flink cdc 1.2.0 时，可以通过指定'debezium.database.serverTimezone' = 'Asia/Shanghai' 来使 timestamp 类型字段显示正常，但 datetime 依旧不正常； 当我升级到 2.0.0 后无法通过 debezium.database.serverTimezone 来改变 timestamp 类型的值

env.addSource(getCDC(DATABASE, TABLE)).print();

mysql中数据：start_time和end_time都是datetime类型
{"id_card":"362324","end_time":2021-08-14 08:55:11,"start_time":2021-08-14 08:44:08,"id":72}

不管是否添加配置：SERVER_TIME_ZONE=Asia/Shanghai，结果都一样
"data":{"id_card":"362324","end_time":1628931311000,"start_time":1628930648000,"id":72}
打印的结果转换为LocalDateTime分别是2021-08-14 16:44:08和2021-08-14 16:55:11

public static DebeziumSourceFunction<JSONObject> getCDC(String databaseList, String tableList) {
    return MySQLSource.<JSONObject>builder()
            .hostname(HOSTNAME)
            .port(3306)
            .username(USERNAME)
            .password(PASSWORD)
            .databaseList(databaseList)
            .tableList(tableList)
            .serverTimeZone(SERVER_TIME_ZONE)
            .deserializer(new MyDeserializer())
            .build();
}

我测试了date，datetime,timestamp,就timestamp增量数据是有问题的，正常datetime是和时区无关的，你应该是配置的问题

[leonardBang]

我测试了各个阶段的数据都没问题的，我在SQL代码里也加了测试： #367
大家会有问题是因为在使用datastream时，你们自定义的序列化器如 MyDeserializer 没有像SQL使用的序列化器RowDataDebeziumDeserializeSchema一样正确地转换timestamp类型，可以参考下RowDataDebeziumDeserializeSchema中对timestamp的处理：

 private TimestampData convertToTimestamp(Object dbzObj, Schema schema) {
        if (dbzObj instanceof Long) {
            switch (schema.name()) {
                case Timestamp.SCHEMA_NAME:
                    return TimestampData.fromEpochMillis((Long) dbzObj);
                case MicroTimestamp.SCHEMA_NAME:
                    long micro = (long) dbzObj;
                    return TimestampData.fromEpochMillis(micro / 1000, (int) (micro % 1000 * 1000));
                case NanoTimestamp.SCHEMA_NAME:
                    long nano = (long) dbzObj;
                    return TimestampData.fromEpochMillis(nano / 1000_000, (int) (nano % 1000_000));
            }
        }
        LocalDateTime localDateTime = TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
        return TimestampData.fromLocalDateTime(localDateTime);
    }

我测试了各个阶段的数据都没问题的，我在SQL代码里也加了测试： #367
大家会有问题是因为在使用datastream时，你们自定义的序列化器如 MyDeserializer 没有像SQL使用的序列化器RowDataDebeziumDeserializeSchema一样正确地转换timestamp类型，可以参考下RowDataDebeziumDeserializeSchema中对timestamp的处理：

 private TimestampData convertToTimestamp(Object dbzObj, Schema schema) {
        if (dbzObj instanceof Long) {
            switch (schema.name()) {
                case Timestamp.SCHEMA_NAME:
                    return TimestampData.fromEpochMillis((Long) dbzObj);
                case MicroTimestamp.SCHEMA_NAME:
                    long micro = (long) dbzObj;
                    return TimestampData.fromEpochMillis(micro / 1000, (int) (micro % 1000 * 1000));
                case NanoTimestamp.SCHEMA_NAME:
                    long nano = (long) dbzObj;
                    return TimestampData.fromEpochMillis(nano / 1000_000, (int) (nano % 1000_000));
            }
        }
        LocalDateTime localDateTime = TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
        return TimestampData.fromLocalDateTime(localDateTime);
    }

我测试的时候用的StringDebeziumDeserializationSchema测试还有自己实现的都是增量binlog的时区和快照时的不一样，请问您是用实现了RowDataDebeziumDeserializeSchema里的这种方式测试的吗?

[leonardBang]

StringDebeziumDeserializationSchema 里没做任何格式的解析， RowDataDebeziumDeserializeSchema里做了，所以你用StringDebeziumDeserializationSchema也需要参考我上面贴的代码

StringDebeziumDeserializationSchema 里没做任何格式的解析， RowDataDebeziumDeserializeSchema里做了，所以你用StringDebeziumDeserializationSchema也需要参考我上面贴的代码

好的，感谢，我测试一下

[luckes-yang]

为什么RowDataDebeziumDeserializeSchema中的convertToTime方法返回的是一个int类型的值，这个怎么转换成java.sql.Time类型呢，我用new java.sql.Time(Long time)这种方式转换的结果比mysql中的实际时间多出8个小时