Debezium 自定义mariadb 连接器的完整代码

以下是使用Debezium自定义MariaDB连接器的完整代码示例：

package com.example.debezium;

import io.debezium.connector.mysql.MySqlConnectorConfig;
import io.debezium.connector.mysql.MySqlConnectorTask;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import io.debezium.relational.history.FileDatabaseHistory;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.storage.StringConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CustomMariaDBConnector {
    private static final Logger LOGGER = LoggerFactory.getLogger(CustomMariaDBConnector.class);

    public static void main(String[] args) throws IOException {
        final String kafkaBootstrapServers = "localhost:9092";
        final String databaseHostName = "localhost";
        final int databasePort = 3306;
        final String databaseUser = "root";
        final String databasePassword = "password";
        final String databaseServerName = "my-server";
        final String databaseHistoryPath = "/path/to/database/history/file";

        final Map<String, String> config = new HashMap<>();
        config.put("name", databaseServerName);
        config.put("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        config.put("database.hostname", databaseHostName);
        config.put("database.port", String.valueOf(databasePort));
        config.put("database.user", databaseUser);
        config.put("database.password", databasePassword);
        config.put("database.server.id", "184054");
        config.put("database.server.name", databaseServerName);
        config.put("database.history", "io.debezium.relational.history.FileDatabaseHistory");
        config.put("database.history.file.filename", databaseHistoryPath);
        config.put("table.whitelist", "mydb.mytable");

        final MySqlConnectorConfig connectorConfig = new MySqlConnectorConfig(config);
        final MySqlConnectorTask task = new MySqlConnectorTask(connectorConfig);
        final JsonConverter keyConverter = new JsonConverter();
        final JsonConverter valueConverter = new JsonConverter();
        final StringConverter headerConverter = new StringConverter();
        keyConverter.configure(config, true);
        valueConverter.configure(config, false);
        headerConverter.configure(config, true);
        final ExecutorService executor = Executors.newSingleThreadExecutor();
        final DebeziumEngine<ChangeEvent<String, String>> engine = EmbeddedEngine.create()
                .using(task)
                .using(keyConverter)
                .using(valueConverter)
                .using(headerConverter)
                .notifying(record -> LOGGER.info("Record: {}", record))
                .using(Json::newBuilder)
                .using(new FileDatabaseHistory())
                .build();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                LOGGER.info("Stopping Debezium engine");
                engine.stop();
                executor.shutdown();
                executor.awaitTermination(10, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                LOGGER.error("Error while stopping Debezium engine", e);
            }
            LOGGER.info("Debezium engine stopped");
        }));

        LOGGER.info("Starting Debezium engine");
        executor.execute(engine);
    }
}

请注意，这个示例是使用Debezium的EmbeddedEngine创建的，但是你也可以使用Debezium的StandaloneEngine。此外，你需要将以下依赖项添加到你的项目中：

- debezium-connector-mysql
- debezium-embedded
- kafka-clients
- kafka-connect-api
- kafka-streams
- slf4j-api

你还需要在你的Kafka集群中创建一个名为“my-server”的主题。