人大金仓数据库V9 mysql兼容版 同步到Doris4.0.2

[ykmx]

1.数据表act_hi_task总数据量180万使用TIS实时同步，S端出现300条左右数据的时间字段为NULL，经过比对发现时间字段NULL值得数据均为较新数据。建表语句如下（人大金仓数据库）

-- `act_hi_taskinst` definition

-- Drop table

-- DROP TABLE `act_hi_taskinst`;

CREATE TABLE `act_hi_taskinst` (
	`ID_` character varying(64) COLLATE `ci_x_icu` NOT NULL,
	`REV_` integer NULL DEFAULT 1,
	`PROC_DEF_ID_` character varying(64) COLLATE `ci_x_icu` NULL,
	`TASK_DEF_ID_` character varying(64) COLLATE `ci_x_icu` NULL,
	`TASK_DEF_KEY_` character varying(255) COLLATE `ci_x_icu` NULL,
	`PROC_INST_ID_` character varying(64) COLLATE `ci_x_icu` NULL,
	`EXECUTION_ID_` character varying(64) COLLATE `ci_x_icu` NULL,
	`SCOPE_ID_` character varying(255) COLLATE `ci_x_icu` NULL,
	`SUB_SCOPE_ID_` character varying(255) COLLATE `ci_x_icu` NULL,
	`SCOPE_TYPE_` character varying(255) COLLATE `ci_x_icu` NULL,
	`SCOPE_DEFINITION_ID_` character varying(255) COLLATE `ci_x_icu` NULL,
	`NAME_` character varying(255) COLLATE `ci_x_icu` NULL,
	`PARENT_TASK_ID_` character varying(64) COLLATE `ci_x_icu` NULL,
	`DESCRIPTION_` character varying(4000) COLLATE `ci_x_icu` NULL,
	`OWNER_` character varying(255) COLLATE `ci_x_icu` NULL,
	`ASSIGNEE_` character varying(255) COLLATE `ci_x_icu` NULL,
	`START_TIME_` datetime(3) NULL,
	`CLAIM_TIME_` datetime(3) NULL,
	`END_TIME_` datetime(3) NULL,
	`DURATION_` bigint NULL,
	`DELETE_REASON_` character varying(4000) COLLATE `ci_x_icu` NULL,
	`PRIORITY_` integer NULL,
	`DUE_DATE_` datetime(3) NULL,
	`FORM_KEY_` character varying(255) COLLATE `ci_x_icu` NULL,
	`CATEGORY_` character varying(255) COLLATE `ci_x_icu` NULL,
	`TENANT_ID_` character varying(255) COLLATE `ci_x_icu` NULL DEFAULT ''::varchar,
	`LAST_UPDATED_TIME_` datetime(3) NULL,
	CONSTRAINT `PRIMARY_idx222` PRIMARY KEY (ID_)
)TABLESPACE sys_default;
CREATE INDEX ACT_IDX_END_TIME_ ON act_hi_taskinst USING btree (END_TIME_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX ACT_IDX_HI_TASK_INST_PROCINST_idx ON act_hi_taskinst USING btree (PROC_INST_ID_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX ACT_IDX_HI_TASK_SCOPE ON act_hi_taskinst USING btree (SCOPE_ID_ NULLS FIRST, SCOPE_TYPE_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX ACT_IDX_HI_TASK_SCOPE_DEF ON act_hi_taskinst USING btree (SCOPE_DEFINITION_ID_ NULLS FIRST, SCOPE_TYPE_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX ACT_IDX_HI_TASK_SUB_SCOPE ON act_hi_taskinst USING btree (SUB_SCOPE_ID_ NULLS FIRST, SCOPE_TYPE_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX idx_PROC_DEF_ID_idx ON act_hi_taskinst USING btree (PROC_DEF_ID_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX idx_assignee_ ON act_hi_taskinst USING btree (ASSIGNEE_ NULLS FIRST) TABLESPACE sys_default;
CREATE INDEX idx_start_time ON act_hi_taskinst USING btree (START_TIME_ NULLS FIRST) TABLESPACE sys_default;

-- insert sql
INSERT INTO act_hi_taskinst (
    ID_, REV_, PROC_DEF_ID_, TASK_DEF_ID_, TASK_DEF_KEY_, PROC_INST_ID_, EXECUTION_ID_,
    SCOPE_ID_, SUB_SCOPE_ID_, SCOPE_TYPE_, SCOPE_DEFINITION_ID_, NAME_, PARENT_TASK_ID_,
    DESCRIPTION_, OWNER_, ASSIGNEE_, START_TIME_, CLAIM_TIME_, END_TIME_, DURATION_,
    DELETE_REASON_, PRIORITY_, DUE_DATE_, FORM_KEY_, CATEGORY_, TENANT_ID_, LAST_UPDATED_TIME_
) VALUES (
    'task12345',                -- ID_ (主键，唯一标识)
    1,                          -- REV_ (乐观锁版本，默认1)
    'procDef:1:12345',          -- PROC_DEF_ID_ (流程定义ID)
    'taskDef123',               -- TASK_DEF_ID_ (任务定义ID)
    'usertask1',                -- TASK_DEF_KEY_ (任务定义KEY)
    'procInst12345',            -- PROC_INST_ID_ (流程实例ID)
    'exec12345',                -- EXECUTION_ID_ (执行实例ID)
    NULL,                       -- SCOPE_ID_
    NULL,                       -- SUB_SCOPE_ID_
    NULL,                       -- SCOPE_TYPE_
    NULL,                       -- SCOPE_DEFINITION_ID_
    '审批任务',                  -- NAME_ (任务名称)
    NULL,                       -- PARENT_TASK_ID_ (父任务ID)
    '这是一个示例任务描述',      -- DESCRIPTION_ (任务描述)
    NULL,                       -- OWNER_ (任务拥有人)
    'user123',                  -- ASSIGNEE_ (任务办理人)
    '2023-01-01 09:00:00.000',  -- 有问题>>START_TIME_ (开始时间)
    NULL,                       -- CLAIM_TIME_ (签收时间)
    '2023-01-02 09:00:00.000',   -- 有问题>> END_TIME_ (结束时间)
    NULL,                       -- DURATION_ (持续时间，单位毫秒)
    NULL,                       -- DELETE_REASON_ (删除原因)
    50,                         -- PRIORITY_ (优先级)
    NULL,                       -- DUE_DATE_ (到期时间)
    NULL,                       -- FORM_KEY_ (表单KEY)
    NULL,                       -- CATEGORY_ (类别)
    '',                         -- TENANT_ID_ (租户ID，默认空字符串)
    '2023-01-01 09:00:00.000'   -- LAST_UPDATED_TIME_ (最后更新时间)
);

2.数据表flow_inst_expand 同步时flink task日志报错，但是数据可以正常同步到doris端
建表语句
-- flow_inst_expand definition

-- Drop table

-- DROP TABLE flow_inst_expand;

CREATE TABLE flow_inst_expand (
id bigint NOT NULL,
proc_inst_id character varying(64) COLLATE ci_x_icu NULL,
main_doc character varying(32) COLLATE ci_x_icu NULL,
document_no character varying(64) COLLATE ci_x_icu NULL,
urgent_flag character varying(2) COLLATE ci_x_icu NULL,
secrecy character varying(32) COLLATE ci_x_icu NULL,
secrecy_term character varying(32) COLLATE ci_x_icu NULL,
storage_period character varying(1000) COLLATE ci_x_icu NULL,
conclude_time_limit datetime NULL,
special_bean_id character varying(64) COLLATE ci_x_icu NULL,
file_system_call_back character varying(255) COLLATE ci_x_icu NULL,
file_system_des character varying(255) COLLATE ci_x_icu NULL,
file_system_file_id character varying(255) COLLATE ci_x_icu NULL,
push_file_system_flag integer NULL,
receive_dept character varying(255) COLLATE ci_x_icu NULL,
form_definition_cache_key character varying(255) COLLATE ci_x_icu NULL,
create_by character varying(255) COLLATE ci_x_icu NULL,
create_time datetime NULL,
update_by character varying(255) COLLATE ci_x_icu NULL,
update_time datetime NULL,
del_flag character varying(2) COLLATE ci_x_icu NULL DEFAULT '0'::varchar,
push_date datetime NULL,
show_color character varying(32) COLLATE ci_x_icu NULL,
CONSTRAINT PRIMARY_idx165 PRIMARY KEY (id),
CONSTRAINT idx_proc_inst_id_idx UNIQUE (proc_inst_id)
)TABLESPACE sys_default;
CREATE INDEX idx_special_bean_id ON flow_inst_expand USING btree (special_bean_id NULLS FIRST) TABLESPACE sys_default;

[baisui1981]

https://github.com/qlangtech/debezium/blob/391dd40e5b4ab4e98d53907f819cf3559cd3a071/debezium-connector-kingbase/src/main/java/io/debezium/connector/postgresql/connection/wal2json/DateTimeFormat.java#L172

按照以下这段代码无论数据库里出来的是什么时间内容都会将他变化为utc时区的instant时间，差8小时的问题就是从这里开始的

        @Override
        public Instant timestampToInstant(String s) {
            return format(TS_FORMAT_PATTERN_HINT, s, () -> LocalDateTime.from(TS_FORMAT.parse(s)).toInstant(ZoneOffset.UTC));
        }

因此，在KingBaseDateTimeConverter 中设置timestampZoneId：

    /**
     * 为了与cdc出来的时间是一致的时区： https://github.com/datavane/tis/issues/492
     * 测试之后发现没有效果
     * props.setProperty("timeZone", ZoneOffset.UTC.getId());
     */
    private final ZoneId timestampZoneId = ZoneOffset.UTC;