Kafka Connect JDBC Sink 连接器 - java.sql.SQLException：找不到合适的驱动程序

Question

我正在尝试在 docker 的帮助下使用 kafka debezium（Kafka 流式传输）将表数据从一个 DB 下沉到另一个 DB。 数据库流工作正常。但是流数据接收另一个 MySQL DB 进程出错。

对于我的连接器接收器配置如下。

 {
  "name": "mysql_sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "mysql-connect.kafka_test.employee",
    "connection.url": "jdbc:mysql://localhost/kafka_test_1&user=debezium&password=xxxxx",
    "auto.create": "true",
    "auto.evolve": "true",
    "insert.mode": "upsert",
    "pk.fields": "id",
    "pk.mode": "record_value",
    "errors.tolerance": "all",
    "errors.log.enable":"true",
    "errors.log.include.messages":"true",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",
    "name": "mysql_sink"
  }
}

但我遇到了一个错误。

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:560)
org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/kafka_test_1&user=debezium&password=xxxxx
io.confluent.connect.jdbc.util.CachedConnectionProvider.getValidConnection(CachedConnectionProvider.java:59)
io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:52)
io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:66)
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)\n\t... 10 more\nCaused by: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/kafka_test_1&user=debezium&password=xxxxx
java.sql.DriverManager.getConnection(DriverManager.java:689)
java.sql.DriverManager.getConnection(DriverManager.java:247)
io.confluent.connect.jdbc.util.CachedConnectionProvider.newConnection(CachedConnectionProvider.java:66)
io.confluent.connect.jdbc.util.CachedConnectionProvider.getValidConnection(CachedConnectionProvider.java:52)\n\t... 13 more

我正在使用 docker。

version: '3'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
     - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    command: [start-kafka.sh]
    ports:
     - "9092:9092"
    links:
     - zookeeper
    environment:
      KAFKA_LISTENERS: PLAINTEXT://:9092,
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on:
      - zookeeper
  connect:
    build:
      context: debezium-jdbc
    ports:
     - "8083:8083"
    links:
     - kafka
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      CLASSPATH: /kafka/connect/kafka-connect-jdbc-5.3.1.jar

我尝试了很多东西，我不知道为什么会出现这个错误，还有一件事我不了解 java。

提前致谢。

Answer 1

在尝试使用 kafka connect 加载 mysql 表时，我一直在努力处理相同的错误 No suitable driver found。

我正在使用 kakfa（不是 confluent 平台），发现您可能会遇到两个问题：

• jdbc url 格式错误
• 为您的 kafka 选择的驱动程序不正确。

我使用了最新的驱动程序mysql-connector-java-8.0.21 并收到没有合适的驱动程序错误。但是，当我切换到版本 mysql-connector-java-5.1.49（2020 年发布）时，一切都像魅力一样。

您可以从 maven repo 获取驱动程序版本： https://mvnrepository.com/artifact/mysql/mysql-connector-java

将驱动程序复制到类路径，在我的情况下，如果下载 kafka 并复制到 kafka_2.12-2.3.1/libs/ 目录中

Answer 2

您收到此错误是因为 JDBCSink（和 JDBCSource）连接器使用 JDBC（顾名思义）连接到数据库，并且您没有使 MySQL 的 JDBC 驱动程序可用于连接器。

解决此问题的最佳方法是将 MySQL JDBC 驱动程序复制到与 kafka-connect-jdbc 相同的文件夹中（在 Docker 映像上为 /usr/share/java/kafka-connect-jdbc/）。

如果您使用的是 Docker Compose，那么您有三个选项。

1. 构建安装了驱动程序的自定义 Docker 映像

2. 在本地下载驱动

   # Download to host machine
   mkdir local-jdbc-drivers
   cd local-jdbc-drivers
   curl https://cdn.mysql.com/Downloads/Connector-J/mysql-connector-java-8.0.18.tar.gz | tar xz 
   

   并将其挂载到容器中到Kafka Connect JDBC的路径中：

   volumes:
     - ${PWD}/local-jdbc-drivers:/usr/share/java/kafka-connect-jdbc/driver-jars/
   

3. 像这样在运行时安装它：

   command: 
     - /bin/bash
     - -c 
     - |
       # JDBC Drivers
       # ------------
       # MySQL
       cd /usr/share/java/kafka-connect-jdbc/
       curl https://cdn.mysql.com/Downloads/Connector-J/mysql-connector-java-8.0.18.tar.gz | tar xz 
       # Now launch Kafka Connect
       sleep infinity &
       /etc/confluent/docker/run 
   

更多详情请见this blog。