kafka JDBC源连接器无法获取postgres表

Posted 2023-03-23

【中文标题】kafka JDBC源连接器无法获取postgres表

【英文标题】：kafka JDBC source connector not able to pick up postgress table

【发布时间】：2021-09-01 01:27:35

【问题描述】：

我有一个在 Windows 上运行的独立 kafka 和一个与我的机器上可用的 postgress db 连接的 JDBC 源连接器。当我尝试运行连接器时，它会显示一条警告，指出不会运行任何任务，因为没有找到表。而如果我将相同的连接器与 mysql db （也是本地安装的）连接。它能够拿起桌子并且工作正常。请注意，我正在使用 confluent 的 jdbc 连接器和默认的 apache-kafka。

以下是我收到的 postgress 警告

这是我的 jdbc source worker 的配置

name=postgress-to-topic
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:postgresql://localhost:5432/test-kafka
connection.user=postgres
connection.password=postgress
topic.prefix=postgres-
#mode=bulk
mode=incrementing
incrementing.column.name=id
errors.log.enable=true
errors.log.include.messages=true
tasks.max=1
poll.interval.ms =100
schema.pattern=test-kafka
#catalog.pattern=test-kafka
table.whitelist=mytable
#dialect.name=PostgreSqlDatabaseDialect
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
topic.prefix=postgress-

这些是同一文件中的 mysql 配置，目前已被注释。

   #mysql configs
    #connection.user=root
    #connection.password=root
    #connection.url=jdbc:mysql://localhost:3306/test-kafka?useSSL=false
    #name=mysql-to-topic
    #topic.prefix=mysql2-
    #schema.pattern=test-kafka
    #table.whitelist=user-data-1

下面是我的 postgress 数据库的样子

下面是我的 mysql 数据库的样子（工作正常）

我对 postgress 没有太多经验，也无法理解为什么它无法在 postgress 中拿起桌子。

【参考方案1】：

相信融合连接器连接器就是debezium连接器

https://www.confluent.io/hub/debezium/debezium-connector-postgresql

这里有文档

https://debezium.io/documentation/reference/connectors/postgresql.html

这需要使用 Postgres 的逻辑复制来获取更改。您需要配置系统以复制要从中获取更改并创建发布的表。

见：

https://www.postgresql.org/docs/10/logical-replication.html

【讨论】：

JDBC Source Connector 与 Debezium 不同

【参考方案2】：

数据库和模式在 postgress 中是两个不同的东西，层次结构如下所示

database-> schema1, schema2
             schema1-> table, table
             schema2-> table, table

我将 schema.pattern 作为导致问题的数据库名称提供，一旦我提供了正确的架构名称，它就可以正常工作。