Kafka Connect 自定义Sink Connector实现在数据同步时增加时间戳字段和假删除功能

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>code.marydon.kafka</groupId>
    <artifactId>marydon-connector-jdbc</artifactId>
    <version>1.1.Final</version>
    <name>kafka-connect-jdbc</name>
    <description>marydon-cdc-jdbc</description>
    <!--项目打包形式-->
    <packaging>jar</packaging>
    <properties>
        <!--maven编译jdk版本-->
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <!--maven插件版本-->
        <maven.compiler.plugin.version>3.8.1</maven.compiler.plugin.version>
        <maven.source.plugin.version>3.8.1</maven.source.plugin.version>
        <!--项目构建字符集-->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <!--<version.debezium>2.3.0.Final</version.debezium>-->
        <!--取决于你所运行的kafka的版本号-->
        <version.kafka>3.5.1</version.kafka>
        <version.debezium>2.5.0.Final</version.debezium>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-api</artifactId>
            <version>${version.kafka}</version>
            <!--仅在编译器有效-->
            <scope>provided</scope>
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-api</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--HikariDataSource是Spring Boot默认的数据库连接池-->
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
            <version>4.0.3</version>
        </dependency>
        <!--mysql驱动-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.28</version>
        </dependency>
        <!--oracle驱动-->
        <dependency>
            <groupId>com.oracle.database.jdbc</groupId>
            <artifactId>ojdbc8</artifactId>
            <version>19.7.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>${project.artifactId}-${project.version}</finalName><!-- 指定package生成的文件名 -->
        <plugins>
            <!--maven项目编译插件-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>${maven.compiler.plugin.version}</version>
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                    <encoding>${project.build.sourceEncoding}</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <executions>
                    <execution>
                        <id>copy-dependencies</id>
                        <phase>package</phase>
                        <goals>
                            <goal>copy-dependencies</goal>
                        </goals>
                        <configuration>
                            <!-- 将依赖的jar包放置在项目构建所在路径的lib目录下 -->
                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
                            <!-- 是否排除间接依赖：间接依赖也拷贝 -->
                            <excludeTransitive>false</excludeTransitive>
                            <!-- 是否跳过版本号：带上版本号 -->
                            <stripVersion>false</stripVersion>
                            <!--排除范围（哪些jar包将被排除在外）-->
                            <excludeScope>provided</excludeScope>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!--maven项目打包插件-->
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

 
/**
 * 输出连接器，用来实现读取配置信息和分配任务等一些初始化工作
 * @description:
 * @author: Marydon
 * @date: 2023-12-06 10:56
 * @version: 1.0
 * @email: [email protected]
 */
@Slf4j
public class JdbcSinkConnector extends SinkConnector {
    private Map<String, String> configProperties;

    // 初始化
    @Override
    public void start(Map<String, String> props) {
        this.configProperties = Map.copyOf(props);

        String topics = props.get(JdbcSinkConfig.TOPICS_REGEX);
        if (StringUtils.isEmpty(topics)) {
            throw new ConnectException("JdbcSinkConnector configuration must include '" + JdbcSinkConfig.TOPICS_REGEX + "' setting");
        }
    }

    //指定要执行的Task类
    @Override
    public Class<? extends Task> taskClass() {
        String driveClassName = configProperties.get(JdbcSinkConfig.CONNECTION_DRIVER_CLASS_NAME);

        // return JdbcSinkTask.class;
        if (driveClassName.indexOf("mysql") > 0)// 目标库是mysql
            return MysqlSinkTask.class;
        else// 目标库是oracle
            return OracleSinkTask.class;
    }

    //task对应的config
    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        log.info("Setting task configurations for {} workers.", maxTasks);
        List<Map<String, String>> configs = new ArrayList<>(maxTasks);

        for(int i = 0; i < maxTasks; i++) {
            configs.add(this.configProperties);
        }

        return configs;
    }

    @Override
    public void stop() {

    }

    //配置定义
    @Override
    public ConfigDef config() {
        // 返回配置定义，用于连接配置的校验
        return JdbcSinkConfig.CONFIG_DEFINITION;
    }

    @Override
    public String version() {
        // return AppInfoParser.getVersion();
        return "1.1.Final";
    }
}

 
/**
 * 数据库动态连接
 * @description: 动态SQL
 * @author: Marydon
 * @date: 2023-12-06 16:06
 * @version: 1.0
 * @email: [email protected]
 */
@Slf4j
public class DynamicConnection {
    private DataSource ds;

    public DynamicConnection(DataSource ds) {
        this.ds = ds;
    }

    public List<Map<String, Object>> queryForList(String sql) throws SQLException {
        Connection conn = null;
        Statement statement = null;
        ResultSet rs = null;
        try {
            conn = this.ds.getConnection();
            statement = conn.createStatement();
            rs = statement.executeQuery(sql);
            return this.rsToList(rs);
        } catch (Exception var10) {
            throw new RuntimeException(var10);
        } finally {
            closeDB(conn, statement, rs, null);
        }
    }

    public List<Map<String, Object>> queryForList(String sql, Object[] params) throws SQLException {
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;

        try {
            conn = this.ds.getConnection();
            ps = conn.prepareStatement(sql);

            for(int i = 0; i < params.length; ++i) {
                ps.setObject(i + 1, params[i]);
            }

            rs = ps.executeQuery();
            return this.rsToList(rs);
        } finally {
            closeDB(conn, ps, rs, null);
        }
    }

    private List<Map<String, Object>> rsToList(ResultSet rs) throws SQLException {
        List<Map<String, Object>> list = new ArrayList<>();
        ResultSetMetaData rsmd = rs.getMetaData();
        int rowCount = rsmd.getColumnCount();
        Map<String, Object> map;

        while(rs.next()) {
            map = new HashMap<>(rowCount);

            for(int i = 1; i <= rowCount; ++i) {
                map.put(rsmd.getColumnName(i), rs.getObject(i));
            }

            list.add(map);
        }

        return list;
    }

    public boolean insert(String sql) throws SQLException {
        Connection conn = null;
        Statement statement = null;
        try {
            conn = this.ds.getConnection();
            statement = conn.createStatement();
            return statement.execute(sql);
        } finally {
            closeDB(conn, statement, null, null);
        }
    }

    public boolean insert(String sql, Object[] params) throws SQLException {
        Connection conn = null;
        PreparedStatement ps = null;
        try {
            conn = this.ds.getConnection();
            ps = conn.prepareStatement(sql);

            for(int i = 0; i < params.length; ++i) {
                ps.setObject(i + 1, params[i]);
            }

            return ps.execute();
        } finally {
            closeDB(conn, ps, null, null);
        }
    }

    public boolean executeBatch(String sql, List<Object[]> list) throws SQLException {
        Connection conn = null;
        PreparedStatement ps = null;
        try {
            conn = this.ds.getConnection();
            conn.setAutoCommit(false);
            ps = conn.prepareStatement(sql);

            for(int i = 0; i < list.size(); ++i) {
                Object[] params = list.get(i);

                for(int j = 0; j < params.length; ++j) {
                    ps.setObject(j + 1, params[j]);
                }

                ps.addBatch();
                if (i % 300 == 0) {
                    ps.executeBatch();
                    ps.clearBatch();
                }
            }

            ps.executeBatch();
            ps.clearBatch();
            conn.commit();
            ps.close();
            return true;
        } catch (Exception var11) {
            if (conn != null) {
                conn.rollback();
            }
            return false;
        } finally {
            closeDB(conn, ps, null, null);
        }

    }

    public int update(String sql) throws SQLException {
        Connection conn = null;
        Statement statement = null;
        try {
            conn = this.ds.getConnection();
            statement = conn.createStatement();
            return statement.executeUpdate(sql);
        } finally {
            closeDB(conn, statement, null, null);
        }
    }

    public int update(String sql, Object[] params) throws SQLException {
        Connection conn = null;
        PreparedStatement ps = null;
        try {
            conn = this.ds.getConnection();
            ps = conn.prepareStatement(sql);

            int result;
            for(result = 0; result < params.length; ++result) {
                ps.setObject(result + 1, params[result]);
            }

            return ps.executeUpdate();
        } finally {
            closeDB(conn, ps, null, null);
        }
    }

    public boolean execProcedure(String sql) throws SQLException {
        return this.execProcedure(sql, new Object[0]);
    }

    public boolean execProcedure(String sql, Object[] params) throws SQLException {
        Connection conn = null;
        CallableStatement cs = null;

        try {
            conn = this.ds.getConnection();
            cs = conn.prepareCall(sql);

            for(int i = 0; i < params.length; ++i) {
                cs.setObject(i + 1, params[i]);
            }

            return cs.execute();
        } finally {
            closeDB(conn, null, null, cs);
        }
    }

    public List<Map<String, Object>> queryProcedure(String sql) throws SQLException {
        return this.queryProcedure(sql, new Object[0]);
    }

    public List<Map<String, Object>> queryProcedure(String sql, Object[] params) throws SQLException {
        Connection conn = null;
        CallableStatement cs = null;
        ResultSet rs = null;

        try {
            conn = this.ds.getConnection();
            log.debug("获取链接时长:");
            cs = conn.prepareCall(sql);
            if (params != null) {
                for(int i = 0; i < params.length; ++i) {
                    cs.setObject(i + 1, params[i]);
                }
            }

            rs = cs.executeQuery();
            log.debug("执行语句时长:,sql:" + sql);
            return this.rsToList(rs);
        } finally {
            log.debug("执行连接池回收");
            closeDB(conn, null, rs, cs);
        }
    }

    public List<Map<String, Object>> queryProcedureMoreRes(String sql) throws SQLException {
        return this.queryProcedure(sql);
    }

    public List<String> getColumnNames(String sql) throws SQLException {
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        List<String> columns = new ArrayList<>();

        try {
            conn = this.ds.getConnection();
            ps = conn.prepareStatement(sql);
            rs = ps.executeQuery();
            ResultSetMetaData rsmd = rs.getMetaData();

            for(int i = 1; i <= rsmd.getColumnCount(); ++i) {
                columns.add(rsmd.getColumnName(i));
            }

            return columns;
        } finally {
            closeDB(conn, ps, rs, null);
        }
    }

    /**
     * 关闭数据库连接
     */
    public void closeDB(Connection conn, Statement ps, ResultSet rs, CallableStatement cs) {
        try {
            if (null != ps && !ps.isClosed()) {
                ps.close();
            }
            if (null != rs && !rs.isClosed()) {
                rs.close();
            }
            if (null != cs && !cs.isClosed()) {
                cs.close();
            }

            if (null != conn && !conn.isClosed()) {
                conn.close();
                log.debug("连接池已关闭");
            }
        } catch (SQLException e) {
            e.printStackTrace();
            log.error("数据库断开连接失败：{}", e.getMessage());
        }
    }
}

/**
 * 动态配置数据源
 * @description:
 * @author: Marydon
 * @date: 2023-12-06 16:03
 * @version: 1.0
 * @email: [email protected]
 */
@Slf4j
public class DynamicDataSource {
    private static final ConcurrentHashMap<String, DataSource> dataSources = new ConcurrentHashMap<>();
    private static DynamicDataSource instance = null;
    private static final Object lock = new Object();
    static int index = 1;

    private DynamicDataSource() {
    }

    private static DynamicDataSource getInstance() {
        if (instance == null) {
            synchronized(DynamicDataSource.class) {
                if (instance == null) {
                    instance = new DynamicDataSource();
                }
            }
        }

        return instance;
    }

    private DataSource getDataSource(String url, String userName, String pwd, String driveClassName, int maxActive) {
        String dataSourceString = url + userName + pwd + driveClassName;
        String hashCode = "ds" + dataSourceString.hashCode();
        if (!dataSources.containsKey(hashCode)) {
            synchronized(lock) {
                if (!dataSources.containsKey(hashCode)) {
                    HikariDataSource ds = new HikariDataSource();
                    ds.setJdbcUrl(url);
                    ds.setUsername(userName);
                    ds.setPassword(pwd);
                    ds.setDriverClassName(driveClassName);
                    ds.setMaximumPoolSize(maxActive);
                    dataSources.put(hashCode, ds);
                    log.debug("DataSource Key:" + hashCode);
                }
            }
        }

        return dataSources.get(hashCode);
    }

    public static DynamicConnection getDynamicConnection(String url, String userName, String pwd, String driveClassName) {
        return getDynamicConnection(url, userName, pwd, driveClassName, 50);
    }

    public static DynamicConnection getDynamicConnection(String url, String userName, String pwd, String driveClassName, int maxActive) {
        DynamicDataSource dynamicDataSource = getInstance();
        DataSource ds = dynamicDataSource.getDataSource(url, userName, pwd, driveClassName, maxActive);
        return new DynamicConnection(ds);
    }

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        ExecutorCompletionService completionService = new ExecutorCompletionService(executorService);

        for(int i = 0; i < 100; ++i) {
            completionService.submit(() ->{
                DynamicConnection conn = code.marydon.db.DynamicDataSource.getDynamicConnection("jdbc:mysql://192.168.0.1:3306/test60", "root", "test", "com.mysql.cj.jdbc.Driver");
                List result = conn.queryForList("select * from base_ac_user limit 1,5");
                Integer[] params = new Integer[]{1, 7};
                DynamicConnection conn1 = code.marydon.db.DynamicDataSource.getDynamicConnection("jdbc:mysql://192.168.0.1:3306/test", "root", "test", "com.mysql.cj.jdbc.Driver");
                List result1 = conn1.queryForList("select * from base_ac_user limit ?,?", params);
                System.out.println(code.marydon.db.DynamicDataSource.index++ + ":" + result.size() + "," + result1.size());
                return null;
            });
        }

    }

}

 /**
 * 接收从kafka订阅到的数据，并将数据同步到mysql库
 * @description: 目标表是mysql数据库
 * @author: Marydon
 * @date: 2023-12-27 11:04
 * @version: 1.0
 * @email: [email protected]
 */
@Slf4j
public class MysqlSinkTask extends SinkTask {
    private DynamicConnection connection;
    // 待同步的表名
    private String tableName = "";
    // 主键字段
    private String primaryFields = "";
    // 获取目标表与源表的字段映射
    String[] targetColumnsMapArray = new String[0];
    // 获取目标表日期字段列
    String[] targetDateColumnsArray = new String[0];
    // 时间戳字段名称
    private String timestampField = "";
    // 删除标识字段名称
    private String deleteMarkField = "";
    // 删除标识字段的值
    private String deleteMarkFieldValue = "";
    // 插入模式
    private String insertMode = "";
    // 日志级别
    private String logLevel = "";
    // 时间戳与真实数据库timestamp字段的时间差（小时差）
    private Integer timestampDifference = 0;

    private String DATE_TIME_FORMAT = "%Y-%m-%d %H:%i:%s";
    private String DATE_FORMAT = "%Y-%m-%d";

    @Override
    public String version() {
        return new JdbcSinkConnector().version();
    }

    //task启动
    @Override
    public void start(Map<String, String> map) {
        String url = map.get(JdbcSinkConfig.CONNECTION_URL);
        String userName = map.get(JdbcSinkConfig.CONNECTION_USER);
        String password = map.get(JdbcSinkConfig.CONNECTION_PASSWORD);
        String driveClassName = map.get(JdbcSinkConfig.CONNECTION_DRIVER_CLASS_NAME);
        log.info("config:{}", map);
        this.connection = DynamicDataSource.getDynamicConnection(url, userName, password, driveClassName);

        tableName = map.get(JdbcSinkConfig.TABLE_NAME_FORMAT);
        primaryFields = map.get(JdbcSinkConfig.PK_FIELDS);
        // 目标表中表示日期的字段
        String targetDateColumns = map.get(JdbcSinkConfig.COLUMNS_DATE);
        // 目标表字段与源表字段的映射关系
        String targetColumnsMap = map.get(JdbcSinkConfig.COLUMNS_MAP);

        targetColumnsMapArray = targetColumnsMap.split(",");
        targetDateColumnsArray = targetDateColumns.split(",");
        timestampField = map.get(JdbcSinkConfig.SYNC_FIELD_TIMESTAMP);
        deleteMarkField = map.get(JdbcSinkConfig.SYNC_FIELD_DELETED);
        deleteMarkFieldValue = map.get(JdbcSinkConfig.SYNC_FIELD_DELETED_MARK);
        insertMode = map.get(JdbcSinkConfig.INSERT_MODE);
        logLevel = map.get(JdbcSinkConfig.LOG_LEVEL);
        String thd = map.get(JdbcSinkConfig.TIMESTAMP_HOURS_DIFFERENCE);
        if (StringUtils.isNotBlank(thd)) {
            timestampDifference = Integer.parseInt(thd);
        }
    }

    //数据put
    @Override
    public void put(Collection<SinkRecord> records) {
        try {
            for (SinkRecord record : records) {
                // 数据处理
                this.pull(record);
            }
        } catch (Exception var4) {
            throw new RuntimeException(var4);
        }
    }

    /**
     * 数据同步（从kafka拉取数据）
     * @explain
     * 1.mysql和Oracle进行互相同步时，必须要保证待同步的字段不包含time类型
     * 2.源表和目标表的主键必须保持一致，因为目标表的数据同步（修改和删除），是按主键走的
     * 3.源表的timestamp类型字段必须与目标表的timestamp类型映射字段保持一致
     * 4.当mysql之间进行数据同步时，支持time类型字段进行同步
     * 5.当mysql之间进行数据同步时，日期类型必须保持一致（datetime-->datetime,date-->date,time-->time,timestamp-->timestamp）
     * @param record
     * @throws SQLException
     */
    private void pull(SinkRecord record) throws SQLException {
        // payload分为：before、after、source、op和ts_ms等几部分
        Struct value = (Struct)record.value();
        if (null == value) return;
        log.debug("待同步数据：{}", value.toString());
        // Struct{after=Struct{patient_id=1, zs_id=2, create_time=1702826679000, update_time=19708, time=2023-12-17T21:24:44Z, id=1}, source=Struct{version=2.2.1.Final, connector=mysql,  name=topic-medi_data_cent-70, ts_ms=1703630644000, snapshot=first, db-me_data_cent, table=t_patient_zs, server_id=0, file=binlog.003268, pos=581446144, row=0}, op=r, ts_ms=1703580582732}
        System.out.println(value);

        // source：事件源的结构信息，包括connector版本、事务ID等
        Struct source = value.getStruct("source");
        // String sourceDatabaseName = source.getString("db");
        // 当入参table.name.format值为""或null时，按目标库的目标表与源库的源表表名一致处理
        if (StringUtils.isEmpty(tableName)) tableName = source.getString("table");

        // 源数据库类型
        String sourceConnectorName = source.getString("connector");
        String operation = value.getString("op");
        // before：变化事件发生之前的值
        Struct beforeStruct = value.getStruct("before");
        // after：变化事件发生之后的值
        Struct afterStruct = value.getStruct("after");

        // READ/CREATE
        if (operation.equals(Operation.READ.code()) || operation.equals(Operation.CREATE.code())) {// r or c
            // 插入操作
            this.insertOperation(afterStruct, sourceConnectorName);
            return;
        }

        // UPDATE
        if (operation.equals(Operation.UPDATE.code())) {// u
            // 当主键字段为空时，禁止更新同步
            if (StringUtils.isEmpty(primaryFields)) return;

            // 执行更新操作
            int updateRows = this.updateOperation(beforeStruct, afterStruct, sourceConnectorName);

            // 更新失败，说明该条记录不存在 && 插入模式为upsert
            if (updateRows < 1 && JdbcSinkConfig.INSERT_MODE_DEFAULT.equalsIgnoreCase(insertMode)) {
                // 不存在，则执行插入操作
                // 说明：oracle必须执行补全日志操作
                this.insertOperation(afterStruct, sourceConnectorName);
            }
            return;
        }

        // DELETE/TRUNCATE
        if (operation.equals(Operation.DELETE.code()) || operation.equals(Operation.TRUNCATE.code())) {// d or t
            // 当主键字段为空时，禁止删除同步
            if (StringUtils.isEmpty(primaryFields)) return;

            // 执行删除操作
            if (StringUtils.isBlank(deleteMarkField)) {
                // 物理删除
                this.physicalDeleteOperation(beforeStruct);
            } else {
                // 逻辑删除
                this.logicalDeleteOperation(beforeStruct);
            }

        }

    }

    /**
     * 插入操作
     * @description: 
     * @date: 2024/3/25 18:29
     * @param afterStruct 变化事件发生之后的值
     * @param sourceConnectorType 源数据库类型
     * @return boolean 插入成功、失败
     */
    private boolean insertOperation(Struct afterStruct, String sourceConnectorType) throws SQLException {
        ArrayList<String> fieldNames = new ArrayList<>();
        ArrayList<String> fieldParams = new ArrayList<>();
        ArrayList<Object> fieldValues = new ArrayList<>();
        String fieldName;
        Object fieldValue;

        // 源字段类型
        String sourceColumn;
        // 目标表字段名称
        String targetColumn;
        // 目标日期字段名称
        String targetDateColumnName;
        // 目标日期字段类型
        String targetDateColumnType;

        for (Field afterField : afterStruct.schema().fields()) {
            fieldName = afterField.name();
            fieldValue = afterStruct.get(fieldName);
            targetColumn = "";

            // 源表与目标表字段映射关系（根据源表字段映射出目标表字段）
            if (targetColumnsMapArray.length > 0) {
                for (String tcm : targetColumnsMapArray) {
                    sourceColumn = tcm.split(":")[0];
                    targetColumn = tcm.split(":")[1];// 关键

                    // 当要新增的列名与此循环的源表列名一致时，结束循环
                    if (sourceColumn.equalsIgnoreCase(fieldName)) break;
                }
            } else {// 说明：入参columns.map没有设值
                // 这样的话，要求目标表字段必须和源表表字段保持一致
                targetColumn = fieldName;
            }

            // 日期字段转换
            // 将日期字段名称，置空
            targetDateColumnName = "";
            targetDateColumnType = "";

            if (targetDateColumnsArray.length > 0) {
                for (String ttc : targetDateColumnsArray) {
                    targetDateColumnName = ttc.split(":")[0];
                    targetDateColumnType = ttc.split(":")[1];

                    // 当要新增的列名为日期字段时，结束循环
                    if (targetDateColumnName.equals(targetColumn)) {
                        break;
                    }

                    // 将日期字段名称，置空
                    targetDateColumnName = "";
                    targetDateColumnType = "";
                }
            } else {// 说明：入参columns.date没有设值
                // 这样的话，要求目标表字段和源表表字段都没有日期字段
                // 将日期字段名称，置空
                targetDateColumnName = "";
                targetDateColumnType = "";
            }

            // 目标日期字段不为空，说明当前要插入的字段是日期类型
            if (!targetDateColumnName.isEmpty()) {
                fieldNames.add(targetColumn);
                boolean isNull = "".equals(fieldValue) || "null".equals(fieldValue) || null == fieldValue;
                if (sourceConnectorType.equals("mysql")) {
                    if (targetDateColumnType.equalsIgnoreCase("datetime")) {// 目标表字段是mysql的datetime字段
                        // mysql datetime
                        // io.debezium.time.Timestamp
                        fieldParams.add("STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT));
                        }
                    } else if (targetDateColumnType.equalsIgnoreCase("date")) {// 目标表字段是mysql的date字段
                        // mysql date
                        // io.debezium.time.Date
                        fieldParams.add("STR_TO_DATE(?, '" + DATE_FORMAT + "')");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.getSomeDay(Integer.parseInt(fieldValue + "")));
                        }
                    } else if (targetDateColumnType.equalsIgnoreCase("timestamp")) {// 目标表字段是mysql的timestamp字段
                        // mysql timestamp
                        // io.debezium.time.ZonedTimestamp
                        fieldParams.add("STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.fromISO8601((String) fieldValue, timestampDifference));
                        }
                    } else {// mysql time
                        // io.debezium.time.MicroTime
                        fieldParams.add("TIME(?)");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.secondsToTime((Long) fieldValue / 1000000));
                        }
                    }
                } else {// 源表oracle
                    if (targetDateColumnType.equalsIgnoreCase("datetime") || targetDateColumnType.equalsIgnoreCase("timestamp")) {// 目标表字段是mysql的datetime字段或者timestamp字段
                        // oracle timestamp-->io.debezium.time.MicroTimestamp
                        // oracle date（包含带时间和不带时间）-->io.debezium.time.Timestamp
                        fieldParams.add("STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT));
                        }
                    } else {// 目标表字段是mysql的date字段
                        // oracle date
                        // io.debezium.time.Timestamp
                        fieldParams.add("STR_TO_DATE(?, '" + DATE_FORMAT + "')");
                        if (isNull) {
                            fieldValues.add(null);
                        } else {
                            fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT).substring(0, 10));
                        }
                    }
                }
            } else {// 非日期字段
                // fieldNames.add(fieldName);
                fieldNames.add(targetColumn);
                fieldParams.add("?");
                fieldValues.add(fieldValue);
            }

            // Date:2024/2/23
            // 增加时间戳标识字段
            if (StringUtils.isNotBlank(timestampField)) {
                fieldNames.add(timestampField);
                fieldParams.add("STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                fieldValues.add(DateUtils.getSysdateStr(null));
            }

        }
        String sql = "insert into " + tableName + "(" + String.join(",", fieldNames) + ")values(" + String.join(",", fieldParams) + ")";

        if (JdbcSinkConfig.LOG_LEVEL_DEFAULT.equalsIgnoreCase(logLevel)) {
            log.info(sql);
        }

        return this.connection.insert(sql, fieldValues.toArray());
    }

    /**
     * 更新操作
     * @description: 
     * @date: 2024/3/25 18:35
     * @param beforeStruct 变化事件发生之前的值
     * @param afterStruct 变化事件发生之后的值
     * @param sourceConnectorType 源数据库类型
     * @return int
     */
    private int updateOperation(Struct beforeStruct, Struct afterStruct, String sourceConnectorType) throws SQLException {
        List<String> fieldNames = new ArrayList<>();
        List<String> fieldParams = new ArrayList<>();
        List<Object> fieldValues = new ArrayList<>();
        String fieldName;
        Object fieldValue;

        // 源字段类型
        String sourceColumn;
        // 目标表字段名称
        String targetColumn;
        // 目标日期字段名称
        String targetDateColumnName;
        // 目标日期字段类型
        String targetDateColumnType;


        // 先比对值前后变化，确定哪些字段需要更新
        // 说明：主键不能变
        for (Field afterField : afterStruct.schema().fields()) {
            fieldName = afterField.name();
            fieldValue = afterStruct.get(fieldName);
            // after和before相同字段的值不一致，说明被更新（mysql，数据更新前后全字段展示，Oracle只提供数据发生变化的字段+主键字段）
            if (!String.valueOf(fieldValue).equals(String.valueOf(beforeStruct.get(fieldName)))) {
                targetColumn = "";
                // 源表与目标表字段映射关系
                for (String tcm : targetColumnsMapArray) {
                    sourceColumn = tcm.split(":")[0];
                    targetColumn = tcm.split(":")[1];

                    // 当要更新的列名与此循环的源表列名一致时，结束循环
                    if (sourceColumn.equalsIgnoreCase(fieldName)) break;
                }

                // 日期字段转换
                // 将日期字段名称，置空
                targetDateColumnName = "";
                targetDateColumnType = "";
                for (String ttc : targetDateColumnsArray) {
                    targetDateColumnName = ttc.split(":")[0];
                    targetDateColumnType = ttc.split(":")[1];

                    // 当要更新的列名为日期字段时，结束循环
                    if (targetDateColumnName.equals(targetColumn)) {
                        break;
                    } else {
                        // 将日期字段名称，置空
                        targetDateColumnName = "";
                        targetDateColumnType = "";
                    }
                }

                // 目标日期字段不为空，说明当前要更新的字段是日期类型
                if (!targetDateColumnName.isEmpty()) {
                    boolean isNull = "".equals(fieldValue) || "null".equals(fieldValue) || null == fieldValue;
                    if (sourceConnectorType.equals("mysql")) {
                        if (targetDateColumnType.equalsIgnoreCase("datetime")) {// 目标表字段是mysql的datetime字段
                            // mysql datetime
                            // io.debezium.time.Timestamp
                            fieldNames.add(targetColumn + "=STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT));
                            }
                        } else if (targetDateColumnType.equalsIgnoreCase("date")) {// 目标表字段是mysql的date字段
                            // mysql date
                            // io.debezium.time.Date
                            fieldNames.add(targetColumn + "=STR_TO_DATE(?, '" + DATE_FORMAT + "')");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.getSomeDay((Integer) fieldValue));
                            }
                        } else if (targetDateColumnType.equalsIgnoreCase("timestamp")) {// 目标表字段是mysql的timestamp字段
                            // mysql timestamp
                            // io.debezium.time.ZonedTimestamp
                            fieldNames.add(targetColumn + "=STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.fromISO8601((String) fieldValue, timestampDifference));
                            }
                        } else {
                            // mysql time
                            // io.debezium.time.MicroTime
                            fieldNames.add(targetColumn + "=TIME(?)");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.secondsToTime((Long) fieldValue / 1000000));
                            }
                        }
                    } else {// 源表oracle
                        if (targetDateColumnType.equalsIgnoreCase("datetime") || targetDateColumnType.equalsIgnoreCase("timestamp")) {// 目标表字段是mysql的datetime字段或者timestamp字段
                            // oracle timestamp-->io.debezium.time.MicroTimestamp
                            // oracle date（包含带时间和不带时间）-->io.debezium.time.Timestamp
                            fieldNames.add(targetColumn + "=STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT));
                            }
                        } else {// 目标表字段是mysql的date字段
                            // oracle date
                            // io.debezium.time.Timestamp
                            fieldNames.add(targetColumn + "=STR_TO_DATE(?, '" + DATE_FORMAT + "')");
                            if (isNull) {
                                fieldValues.add(null);
                            } else {
                                fieldValues.add(DateUtils.timestampToString((Long) fieldValue, JdbcSinkConfig.DB_TIMEZONE_DEFAULT).substring(0, 10));
                            }
                        }
                    }
                } else {// 非日期字段
                    // fieldNames.add(fieldName + "=?");
                    fieldNames.add(targetColumn + "=?");
                    fieldValues.add(fieldValue);
                }
            }

            // 增加时间戳标识字段
            if (StringUtils.isNotBlank(timestampField)) {
                fieldNames.add(timestampField + "=STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
                fieldValues.add(DateUtils.getSysdateStr(null));
            }
        }

        // where条件，根据主键更新
        for (String pk : primaryFields.split(",")) {
            fieldParams.add(pk + "=?");
            for (Field f : afterStruct.schema().fields()) {
                fieldName = f.name();

                if (StringUtils.equalsIgnoreCase(fieldName, pk)) {
                    fieldValues.add(afterStruct.get(pk));
                    break;
                }
            }

        }

        String sql = "UPDATE " + tableName + " SET " + String.join(",", fieldNames) + " WHERE " + String.join(" AND ", fieldParams);

        if (JdbcSinkConfig.LOG_LEVEL_DEFAULT.equalsIgnoreCase(logLevel)) {
            log.info(sql);
        }

        // 执行更新操作
        return this.connection.update(sql, fieldValues.toArray());
    }

    /**
     * 删除操作（物理删除）
     * @description:
     * @date: 2024/3/25 18:35
     * @param beforeStruct 变化事件发生之前的值
     * @return int 删除的记录数
     */
    private int physicalDeleteOperation(Struct beforeStruct) throws SQLException {
        ArrayList<String> fieldParams = new ArrayList<>();
        ArrayList<Object> fieldValues = new ArrayList<>();
        String fieldName;

        // where条件，根据主键删除
        for (String pk : primaryFields.split(",")) {
            fieldParams.add(pk + "=?");
            for (Field f : beforeStruct.schema().fields()) {
                fieldName = f.name();

                if (StringUtils.equalsIgnoreCase(fieldName, pk)) {
                    fieldValues.add(beforeStruct.get(pk));
                    break;
                }
            }

        }

        // 根据主键删除记录
        String sql = "DELETE FROM " + tableName + " WHERE " + String.join(" AND ", fieldParams);

        if (JdbcSinkConfig.LOG_LEVEL_DEFAULT.equalsIgnoreCase(logLevel)) {
            log.info(sql);
        }

        return this.connection.update(sql, fieldValues.toArray());
    }

    /**
     * 删除操作（逻辑删除、假删除）
     * @description:
     * @date: 2024/3/25 18:35
     * @param beforeStruct 变化事件发生之前的值
     * @return int 删除的记录数
     */
    private int logicalDeleteOperation(Struct beforeStruct) throws SQLException {
        List<String> fieldNames = new ArrayList<>();
        List<String> fieldParams = new ArrayList<>();
        List<Object> fieldValues = new ArrayList<>();
        String fieldName;

        // Date:2024/2/23
        // 增加标识是否删除字段
        fieldNames.add(deleteMarkField + "=?");
        fieldValues.add(deleteMarkFieldValue);

        // 增加时间戳标识字段
        if (StringUtils.isNotBlank(timestampField)) {
            fieldNames.add(timestampField + "=STR_TO_DATE(?, '" + DATE_TIME_FORMAT + "')");
            fieldValues.add(DateUtils.getSysdateStr(null));
        }

        // where条件，根据主键更新
        for (String pk : primaryFields.split(",")) {
            fieldParams.add(pk + "=?");
            for (Field f : beforeStruct.schema().fields()) {
                fieldName = f.name();

                if (StringUtils.equalsIgnoreCase(fieldName, pk)) {
                    fieldValues.add(beforeStruct.get(pk));
                    break;
                }
            }

        }

        String sql = "UPDATE " + tableName + " SET " + String.join(",", fieldNames) + " WHERE " + String.join(" AND ", fieldParams);

        if (JdbcSinkConfig.LOG_LEVEL_DEFAULT.equalsIgnoreCase(logLevel)) {
            log.info(sql);
        }

        return this.connection.update(sql, fieldValues.toArray());
    }
    @Override
    public void stop() {

    }
}

 /**
 * 日期工具类
 * https://www.cnblogs.com/Marydon20170307/p/17921900.html
 * @description: 处理日期间的格式转换
 * @explain:
 * mysql的datetime类型会被插件io.debezium.connector.mysql.MySqlConnector转成时间戳，需调用DateUtils.timestampToString(1702027934000L,"UTC")
 * mysql的date类型会被插件io.debezium.connector.mysql.MySqlConnector转成天数，需调用getSomeDay(19699)
 * mysql的time类型会被插件io.debezium.connector.mysql.MySqlConnector转成微秒数，需调用DateUtils.secondsToTime(34419000000L / 1000000)
 * mysql的timestamp类型会被插件io.debezium.connector.mysql.MySqlConnector转成ISO8601格式的时间戳，需调用DateUtils.fromISO8601("2023-12-08T15:32:19Z")
 * oracle的date类型和timestamp类型会被插件io.debezium.connector.oracle.OracleConnector转成时间戳，需调用timestampToString(1702837490000L,"UTC")，timestampToString(1702826697000000L,"UTC")
 * 从转换结果来看，oracle的timestamp类型被插件转换后只是被date类型多了3个0
 * @author: Marydon
 * @date: 2023-12-07 11:48
 * @version: 1.0
 * @email: [email protected]
 */
public class DateUtils {
    // 年月日（大写M：表示月份，小写m：表示分钟）
    public static final String FORMAT_DATE = "yyyy-MM-dd";
    // 时分秒（大写H：表示24小时制，小写h：表示12小时制）
    public static final String FORMAT_TIME = "HH:mm:ss";
    // 年月日时分秒
    public static final String FORMAT_DATE_TIME = FORMAT_DATE + " " + FORMAT_TIME;
    // 时区
    // 东八区：GMT+8/UTC+8
    public static final String FORMAT_TIME_ZONE = "Asia/Shanghai";

    public static final Integer DEFAULT_DIFFERENCE_HOURS = -6;

    /**
     * 将时间戳转成日期字符串
     * @param timestamp 时间
     * @return 日期格式字符串
     */
    public static String timestampToString(Long timestamp) {
        return toDateTimeString(fromTimeMills(timestamp), FORMAT_DATE_TIME);
    }

    public static String timestampToString(Long timestamp, String timeZone) {
        if (String.valueOf(timestamp).length() == 16) {// 16：毫秒
            return toDateTimeString(fromTimeMills(timestamp / 1000, timeZone), FORMAT_DATE_TIME);
        } else {// 13：秒
            return toDateTimeString(fromTimeMills(timestamp, timeZone), FORMAT_DATE_TIME);
        }
    }

    /**
     * 日期转字符串（日期+时间）
     * @attention: jdk>=1.8
     * @date: 2020年08月31日 0031 17:04
     * @param: dateTime
     * @param: pattern
     * @return: java.lang.String
     */
    public static String toDateTimeString(LocalDateTime dateTime, String pattern) {
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(pattern);
        return dateTimeFormatter.format(dateTime);
    }

    /**
     * 毫秒数转LocalDateTime
     * @attention: jdk>=1.8
     * @date: 2020年09月10日 0010 11:22
     * @param: timeMills
     * @return: java.time.LocalDateTime
     */
    public static LocalDateTime fromTimeMills(long timeMills){
        return fromTimeMills(timeMills, FORMAT_TIME_ZONE);
    }

    public static LocalDateTime fromTimeMills(long timeMills, String timeZone){
        return LocalDateTime.ofInstant(Instant.ofEpochMilli(timeMills), ZoneId.of(timeZone));
    }

    /**
     * 日期字符串按指定格式转LocalDateTime
     * @attention:
     * @date: 2021/7/28 15:05
     * @param: dateTimeStr 日期字符串
     * 2023-12-07T16:00:00Z
     * "2023-12-07T16:00:00Z"是一种ISO 8601标准的日期时间表示方式
     * 这个字符串表示的是一个特定的时间点：2023年12月7日，下午4点（16点），0分钟，0秒。其中“T”是时间标识符，“Z”表示的是协调世界时（UTC）。
     * 这种格式是可以精确到秒的时间戳
     * @return: java.time.LocalDateTime
     */
    public static LocalDateTime toLocalDateTime(String dateTimeStr) {
        // UTC-->这个位置不知道debezium是怎么处理的，按理说该是相差8个小时才对。但实际上相差了6个小时
        return toLocalDateTime(dateTimeStr, DEFAULT_DIFFERENCE_HOURS);
    }

    public static LocalDateTime toLocalDateTime(String dateTimeStr, Integer hours) {
        // UTC时间
        DateTimeFormatter formatter = DateTimeFormatter.ISO_DATE_TIME;
        ZonedDateTime zonedDateTime = ZonedDateTime.parse(dateTimeStr, formatter);

        // // 使用DateTimeFormatter.ISO_INSTANT解析字符串为Instant对象
        // DateTimeFormatter formatter = DateTimeFormatter.ISO_INSTANT;
        // Instant instant = Instant.from(formatter.parse(dateTimeStr));
        // // 将Instant对象转换为LocalDateTime
        // return instant.atZone(ZoneId.of("UTC")).toLocalDateTime();

        return zonedDateTime.toLocalDateTime().plusHours(hours);
    }

    /**
     * ISO 8601标准日期转成字符串
     * @param dateTimeStr
     * 2023-12-07T16:00:00Z
     * "2023-12-07T16:00:00Z"是一种ISO 8601标准的日期时间表示方式
     * 这个字符串表示的是一个特定的时间点：2023年12月7日，下午4点（16点），0分钟，0秒。其中“T”是时间标识符，“Z”表示的是协调世界时（UTC）。
     * 这种格式是可以精确到秒的时间戳
     * @return
     */
    public static String fromISO8601(String dateTimeStr) {
        return toDateTimeString(toLocalDateTime(dateTimeStr), FORMAT_DATE_TIME);
    }
    public static String fromISO8601(String dateTimeStr, Integer hours) {
        return toDateTimeString(toLocalDateTime(dateTimeStr, hours), FORMAT_DATE_TIME);
    }

    /**
     * 根据天数倒推日期
     * https://www.cnblogs.com/Marydon20170307/p/10672030.html
     * @param days 距离1970-01-01的天数
     * @return yyyy-MM-dd
     * 2023-12-26
     */
    public static String getSomeDay(int days) {
        SimpleDateFormat sdf = new SimpleDateFormat(FORMAT_DATE);
        Date date;
        try {
            date = sdf.parse("1970-01-01");
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(date);
        calendar.add(Calendar.DAY_OF_YEAR, days);
        date = calendar.getTime();
        return sdf.format(date);
    }

    /**
     * 根据秒数倒推时间
     * @param seconds
     * @return HH:mm:ss
     * 09:30:00
     * 21:30:00
     */
    public static String secondsToTime(Long seconds) {
        String dateTime = String.format("%d:%d:%d", seconds / 3600, (seconds % 3600) / 60, seconds % 60);
        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
        Date date;
        try {
            date = sdf.parse(dateTime);
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        return sdf.format(date);
    }

    /**
     * 获取系统当前时间
     * @return 指定格式的系统时间
     */
    public static String getSysdateStr(String format) {
        // format = format == null || format == "" ? FORMAT_DATE_TIME : format;
        format = Optional.ofNullable(format).filter(str -> !str.isEmpty()).orElse(FORMAT_DATE_TIME);

        LocalDateTime now = LocalDateTime.now();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern(format);
        return now.format(formatter);
    }

    public static void main(String[] args) {
        // mysql
        // System.out.println(timestampToString(1702027934000L,"UTC"));
        System.out.println(fromISO8601("2023-12-08T15:32:19Z"));
        // System.out.println(getSomeDay(19699));
        // System.out.println(secondsToTime(34419000000L / 1000000));
        // oracle
        // System.out.println(timestampToString(1702837490000L,"UTC"));
        // System.out.println(timestampToString(1702826697000000L,"UTC"));
        // System.out.println(timestampToString(1708330724413L,"UTC+8"));
        // System.out.println(secondsToTime(77400L));

    }
}