Kafka JDBC Sink连接器启动失败：方言识别异常及配置问题求助

问题背景 #

启动Confluent Kafka JDBC Sink连接器，将Kafka主题account-operations的数据写入PostgreSQL数据库时遇到异常，尝试指定方言名称后仍报错。

连接器配置 #

{
  "name": "postgres-balance-sink-connector-name",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "account-operations",
    "name": "postgres-balance-sink-connector-name",
    "connection.url": "jdbc:postgresql://db:5432/kafka_db",
    "connection.user": "postgres",
    "connection.password": "example",
    "insert.mode": "UPSERT",
    "auto.create": "true",
    "auto.evolve": "true",
    "tasks.max": "1"
  }
}

初始启动错误日志 #

[2023-09-21 18:29:37,312] ERROR [postgres-balance-sink-connector-name|task-0] WorkerSinkTask{id=postgres-balance-sink-connector-name-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:210)
2023-09-21T18:29:37.313123031Z java.lang.NullPointerException: Cannot invoke "io.confluent.connect.jdbc.dialect.DatabaseDialectProvider.create(org.apache.kafka.common.config.AbstractConfig)" because "bestMatch" is null
2023-09-21T18:29:37.313126905Z  at io.confluent.connect.jdbc.dialect.DatabaseDialects.findBestFor(DatabaseDialects.java:134)
2023-09-21T18:29:37.313130191Z  at io.confluent.connect.jdbc.sink.JdbcSinkTask.initWriter(JdbcSinkTask.java:68)
2023-09-21T18:29:37.313133312Z  at io.confluent.connect.jdbc.sink.JdbcSinkTask.start(JdbcSinkTask.java:53)
2023-09-21T18:29:37.313136366Z  at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:315)
2023-09-21T18:29:37.313139418Z  at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:200)
2023-09-21T18:29:37.313142452Z  at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
2023-09-21T18:29:37.313145341Z  at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
2023-09-21T18:29:37.313148405Z  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
2023-09-21T18:29:37.313151410Z  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
2023-09-21T18:29:37.313154545Z  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
2023-09-21T18:29:37.313157629Z  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
2023-09-21T18:29:37.313160698Z  at java.base/java.lang.Thread.run(Thread.java:833)
2023-09-21T18:29:37.313211156Z [2023-09-21 18:29:37,313] INFO [postgres-balance-sink-connector-name|task-0] Stopping task (io.confluent.connect.jdbc.sink.JdbcSinkTask:170)
2023-09-21T18:29:37.313924296Z [2023-09-21 18:29:37,313] WARN [postgres-balance-sink-connector-name|task-0] Could not stop task (org.apache.kafka.connect.runtime.WorkerSinkTask:175)
2023-09-21T18:29:37.313937766Z java.lang.NullPointerException: Cannot invoke "io.confluent.connect.jdbc.sink.JdbcDbWriter.closeQuietly()" because "this.writer" is null
2023-09-21T18:29:37.313941811Z  at io.confluent.connect.jdbc.sink.JdbcSinkTask.stop(JdbcSinkTask.java:172)
2023-09-21T18:29:37.313958175Z  at org.apache.kafka.connect.runtime.WorkerSinkTask.close(WorkerSinkTask.java:173)
2023-09-21T18:29:37.313961419Z  at org.apache.kafka.connect.runtime.WorkerTask.doClose(WorkerTask.java:181)
2023-09-21T18:29:37.313964049Z  at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:214)
2023-09-21T18:29:37.313966924Z  at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
2023-09-21T18:29:37.313969524Z  at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
2023-09-21T18:29:37.313972291Z  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
2023-09-21T18:29:37.313975315Z  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
2023-09-21T18:29:37.313978394Z  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
2023-09-21T18:29:37.313981380Z  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
2023-09-21T18:29:37.313984474Z  at java.base/java.lang.Thread.run(Thread.java:833)

尝试添加方言配置后的错误 #

添加如下配置：

"dialect.name": "PostgreSqlDatabaseDialect",

得到错误：

{"error_code":400,"message":"Connector configuration is invalid and contains the following 1 error(s):\nInvalid value PostgreSqlDatabaseDialect for configuration dialect.name: Invalid enumerator\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"}

解决建议 #

• 检查PostgreSQL JDBC驱动是否存在：Confluent JDBC连接器本身不包含数据库驱动，需将PostgreSQL的JDBC驱动jar包（如postgresql-42.6.0.jar）放置在Kafka Connect的插件目录（通常是share/java/kafka-connect-jdbc/）下，重启Connect服务后重试。驱动缺失会导致连接器无法识别数据库类型，抛出找不到方言的NPE。
• 核对连接器版本与文档版本：不同版本的JDBC连接器对dialect.name的枚举值定义可能不同。若使用旧版本，尝试使用简写值PostgreSQL而非全称；若版本匹配仍报错，调用Connect的配置验证接口/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate，查看该配置项允许的有效值列表。
• 尝试使用全限定类名：部分版本可能需要指定方言的完整类名作为dialect.name的值，即io.confluent.connect.jdbc.dialect.PostgreSqlDatabaseDialect，替换原配置值后测试。
• 验证数据库连接可用性：确认PostgreSQL服务正常运行，connection.url中的主机、端口、库名正确，connection.user和connection.password具备目标数据库的读写权限，排除基础连接问题。

内容的提问来源于stack exchange，提问作者papaj