mysql数据发送到kafka

import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.storage.FileOffsetStorage;
import org.apache.kafka.connect.storage.OffsetStorageReader;

public class MySQLSourceConnector extends SourceConnector {
    @Override
    public Class<? extends SourceTask> taskClass() {
        return MySQLSourceTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        List<Map<String, String>> configs = new ArrayList<>();
        for (int i = 0; i < maxTasks; i++) {
            Map<String, String> config = new HashMap<>();
            config.put("connector.class", "io.confluent.connect.jdbc.JdbcSourceConnector");
            config.put("tasks.max", "1");
            config.put("connection.url", "jdbc:mysql://localhost:3306/mydb");
            config.put("table.whitelist", "mytable");
            config.put("mode", "timestamp");
            config.put("timestamp.column.name", "updated_at");
            config.put("topic.prefix", "mytable_");
            configs.add(config);
        }
        return configs;
    }

    @Override
    public void start(Map<String, String> props) {
        // 初始化连接器
    }

    @Override
    public void stop() {
        // 停止连接器
    }

    @Override
    public ConfigDef config() {
        return new ConfigDef();
    }
}

public class MySQLSourceTask extends SourceTask {
    private OffsetStorageReader offsetStorageReader;

    @Override
    public void start(Map<String, String> props) {
        // 初始化任务
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        List<SourceRecord> records = new ArrayList<>();
        // 从MySQL读取数据并转换为SourceRecord
        return records;
    }

    @Override
    public void commit() throws InterruptedException {
        // 提交偏移量
    }

    @Override
    public void stop() {
        // 停止任务
    }
}