Confluent社区版创建JDBC源连接器报错：找不到对应Connector类

安装Confluent Platform社区版7.0.1（tar包）后，在ksql中执行创建PostgreSQL JDBC源连接器的命令时，报错无法找到io.confluent.connect.jdbc.JdbcSourceConnector类，当前可用连接器列表中也没有该类。

执行的命令：

ksql> CREATE SOURCE CONNECTOR `postgres-jdbc-source` WITH (
>   "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
>   "connection.url"='jdbc:postgresql://postgres:5432/postgres',
>   "mode"='incrementing',
>   "incrementing.column.name"='ref',
>   "table.whitelist"='carusers',
>   "connection.password"='postgres',
>   "connection.user"='postgres',
>   "topic.prefix"='db-',
>   "key"='username');

错误信息：

{
  "error_code" : 500,
  "message" : "Failed to find any class that implements Connector and which name matches io.confluent.connect.jdbc.JdbcSourceConnector, available connectors are: PluginDesc{klass=class io.confluent.connect.replicator.ReplicatorSourceConnector, name='io.confluent.connect.replicator.ReplicatorSourceConnector', version='7.0.1', encodedVersion=7.0.1, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/kafka-connect-replicator/'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorCheckpointConnector, name='org.apache.kafka.connect.mirror.MirrorCheckpointConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorHeartbeatConnector, name='org.apache.kafka.connect.mirror.MirrorHeartbeatConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorSourceConnector, name='org.apache.kafka.connect.mirror.MirrorSourceConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockConnector, name='org.apache.kafka.connect.tools.MockConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=connector, typeName='connector', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSinkConnector, name='org.apache.kafka.connect.tools.MockSinkConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=sink, typeName='sink', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSourceConnector, name='org.apache.kafka.connect.tools.MockSourceConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.SchemaSourceConnector, name='org.apache.kafka.connect.tools.SchemaSourceConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSinkConnector, name='org.apache.kafka.connect.tools.VerifiableSinkConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSourceConnector, name='org.apache.kafka.connect.tools.VerifiableSourceConnector', version='7.0.1-ce', encodedVersion=7.0.1-ce, type=source, typeName='source', location='file:/Users/prats/confluent-7.0.1/share/java/confluent-kafka-mqtt/'}"
} 

Confluent社区版tar包默认不包含JDBC连接器插件，需手动安装：

1. 下载对应版本的JDBC连接器插件
   获取与Confluent 7.0.1版本严格匹配的kafka-connect-jdbc插件包，版本不一致会导致兼容性问题。

2. 安装插件到Kafka Connect的插件目录
   在Confluent安装目录的share/java文件夹下创建kafka-connect-jdbc目录，将下载的插件解压或复制到该目录。示例路径：

   /Users/prats/confluent-7.0.1/share/java/kafka-connect-jdbc/
   

   注意：若插件包未包含PostgreSQL的JDBC驱动，需单独下载对应版本的postgresql-*.jar并放入该目录。

3. 重启Kafka Connect服务
   Kafka Connect仅在启动时加载插件，必须重启服务才能识别新安装的JDBC连接器。

4. 验证插件加载状态
   通过以下命令查看已加载的连接器插件，确认io.confluent.connect.jdbc.JdbcSourceConnector已在列表中：

   curl http://localhost:8083/connector-plugins
   

5. 重新执行创建连接器命令
   再次运行之前的ksql创建命令，即可正常创建PostgreSQL JDBC源连接器。

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