Kafka Connect JDBC 源连接器不适用于 Microsoft SQL Server

Question

我已经设置了一个 Kafka Connect 的 dockerized 集群，它以分布式模式运行。 我正在尝试设置一个 Kafka JDBC 源连接器来在 Microsoft SQL Server 和 Kafka 之间移动数据。

下面是我的connector-plugins api 的响应输出

[
    {
    class: "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    type: "sink",
    version: "4.0.0"
    },
    {
    class: "io.confluent.connect.hdfs.HdfsSinkConnector",
    type: "sink",
    version: "4.0.0"
    },
    {
    class: "io.confluent.connect.hdfs.tools.SchemaSourceConnector",
    type: "source",
    version: "1.0.0-cp1"
    },
    {
    class: "io.confluent.connect.jdbc.JdbcSinkConnector",
    type: "sink",
    version: "4.0.0"
    },
    {
    class: "io.confluent.connect.jdbc.JdbcSourceConnector",
    type: "source",
    version: "4.0.0"
    },
    {
    class: "io.debezium.connector.mongodb.MongoDbConnector",
    type: "source",
    version: "0.7.4"
    },
    {
    class: "io.debezium.connector.mysql.MySqlConnector",
    type: "source",
    version: "0.7.4"
    },
    {
    class: "org.apache.kafka.connect.file.FileStreamSinkConnector",
    type: "sink",
    version: "1.0.0-cp1"
    },
    {
    class: "org.apache.kafka.connect.file.FileStreamSourceConnector",
    type: "source",
    version: "1.0.0-cp1"
    }
]

我已经将JDBC Driver provided my Microsoft SQL Server 添加到我的Kafka Connect 集群中的plugins path。

下面是我的connectors api 的输入，

curl -X POST \
  http://kafka-connect-cluster.com/connectors \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -d '{
"name": "mssql-source-connector",
"config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "mode": "timestamp",
        "timestamp.column.name": "updateTimeStamp",
        "query": "select * from table_name",
        "tasks.max": "1",
        "table.types": "TABLE",
        "key.converter.schemas.enable": "false",
        "topic.prefix": "data_",
        "value.converter.schemas.enable": "false",
        "connection.url": "jdbc:sqlserver://<host>:<port>;databaseName=<dbName>;",
        "connection.user": "<username>",
        "connection.password": "<password>",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "poll.interval.ms": "5000",
        "table.poll.interval.ms": "120000"
    }
}'

我在尝试此查询时遇到的错误如下：

{
    "error_code": 400,
    "message": "Connector configuration is invalid and contains the following 2 error(s):\nInvalid value java.sql.SQLException: No suitable driver found for jdbc:sqlserver://<host>:<port>;databaseName=<db_name>; for configuration Couldn't open connection to jdbc:sqlserver://<host>:<port>;databaseName=<db_name>;\nInvalid value java.sql.SQLException: No suitable driver found for jdbc:sqlserver://<host>:<port>;databaseName=<db_name;> for configuration Couldn't open connection to jdbc:sqlserver://<host>:<port>;databaseName=<db_name;>\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"
}

我们非常感谢您提供的任何帮助。

谢谢

Answer 1

感谢@rmoff 为我指明了正确的方向。

所以问题出在两个地方。

1. 这更像是一个仅供参考，而不是一个问题。我给了 docker 镜像一个自定义的CONNECT_PLUGIN_PATH。有 这样做没有错，但通常不是一个好主意 因为您必须复制所有可用的基本插件 融合平台，这可能会在迁移到新平台时产生问题 版本，因为您可能必须再次执行相同的过程。
2. 这部分是最重要的。 SQLServer JDBC driver 需要 与 kafka-connect-jdbc-<confluent-version>.jar 在我的情况下是 kafka-connect-jdbc-4.0.0.jar。

解决这两点后，我的 SQLServer JDBC 驱动程序开始按预期工作。

Answer 2

基于https://docs.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url，您在 URL 中的结尾 ; 无效。 还可以尝试将 JDBC 驱动程序放在 share/java/kafka-connect-jdbc 中，和/或将其添加到 CLASSPATH 环境变量中。