sql-server - 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给我的plugins path在我的 Kafka Connect 集群中。

以下是我的 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 为我指明了正确的方向。

所以问题出在两个地方。

这更像是一个仅供引用，而不是一个问题。我给了 docker 镜像一个自定义的 CONNECT_PLUGIN_PATH .有
这样做没有错，但通常不是一个好主意
因为您必须复制所有可用的基本插件
融合平台，这可能会在迁移到新平台时产生问题
版本，因为您可能必须再次经历相同的过程。

这部分是最重要的。 SQLServer JDBC driver需要
与kafka-connect-jdbc-<confluent-version>.jar在我的情况下是kafka-connect-jdbc-4.0.0.jar .

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