Kafka Connect无法加载自定义JdbcSinkConnector问题求助

你的自定义JdbcSinkConnector已被Kafka Connect加载注册，但未被识别为可用连接器，核心原因是Kafka Connect无法自动发现你的连接器实现类，同时代码存在逻辑缺陷，以下是具体解决步骤：

1. 添加服务注册文件（关键） #

Kafka Connect通过META-INF/services/org.apache.kafka.connect.connector.Connector文件自动识别连接器实现类，你的Jar包大概率缺少该文件：

• 在项目src/main/resources/META-INF/services/目录下创建文件org.apache.kafka.connect.connector.Connector
• 文件内写入自定义Connector的全限定类名，例如：

  com.yourpackage.CustomJdbcSinkConnector
  

• 重新打包Jar，用命令jar tf custom-jdbc-sink-1.0.0.jar验证，输出需包含META-INF/services/org.apache.kafka.connect.connector.Connector路径。

2. 完善CustomJdbcSinkConnector实现 #

当前Connector类仅重写了taskClass()和version()，缺少必要的配置声明与初始化逻辑，补充如下：

public class CustomJdbcSinkConnector extends JdbcSinkConnector {

    @Override
    public Class<? extends Task> taskClass() {
        return CustomJdbcSinkTask.class;
    }

    @Override
    public String version() {
        return "1.0.0";
    }

    // 返回JDBC Sink的配置定义，可添加自定义配置
    @Override
    public ConfigDef config() {
        return JdbcSinkConfig.configDef();
    }

    // 确保父类初始化逻辑被执行
    @Override
    public void start(Map<String, String> props) {
        super.start(props);
    }
}

3. 修复CustomJdbcSinkTask的put方法逻辑 #

你当前在put()中新建的JdbcSinkTask实例未经过Connect初始化流程（无配置、数据库连接等），会导致执行失败，应调用父类方法：

public class CustomJdbcSinkTask extends JdbcSinkTask {

    private static final Logger log = LoggerFactory.getLogger(CustomJdbcSinkTask.class);

    @Override
    public void put(Collection<SinkRecord> records) {
        log.info("hello from custom put");
        log.info("custom put method starting...");
        // 复用父类的JDBC写入逻辑
        super.put(records);
    }
}

4. 验证插件加载 #

重启Kafka Connect后，通过REST API查看已加载的连接器：

curl http://localhost:8083/connector-plugins

若自定义连接器出现在返回列表中，说明加载成功，即可创建对应的连接器实例。

内容的提问来源于stack exchange，提问作者Muhamed Risvan M S