apache-kafka - Kafka Connect JDBC 接收器连接器 - java.sql.SQLException : No suitable driver found

Question

我正在尝试在 docker 的帮助下使用 kafka debezium(Kafka 流)将一个数据库的表数据下沉到另一个数据库。
数据库流工作正常。但是流式传输数据以接收另一个 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

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

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

如果您使用 Docker Compose，那么您有三个选择。

使用已安装的驱动程序构建自定义 Docker 镜像

本地下载驱动

# 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/

在运行时安装它，如下所示:

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 .