Kafka 与 Debezium 连接答案

【问题标题】：Kafka Connect with DebeziumKafka 与 Debezium 连接
【发布时间】：2019-01-17 03:37:38
【问题描述】：

我已经使用 Postgresql 数据库设置了环境，将 debezium 连接器与 Kafka Connect 和 Kafka 一起使用。有多个 Kafka 实例（3）在运行，它配置了 Zookeeper（3）。整个管道中的连接都在工作，但是根据 Debezium 的文档，没有按照表中的表自动创建主题数据库。例如，如果是某个模式中的表 A 和表 B，我假设在 Kafka 中隐式创建了 2 个主题。连接器和任务的状态是RUNNING，下面是我为连接器所做的配置，

{
   "name": "geo-connector",
   "config": {
    "connector.class":     "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": <dbHostName>,
    "database.port": <dbPort>,
    "database.user": <dbUser>,
    "database.password":<dbPassword> ,
    "database.dbname" : <dbName>,
    "database.server.name": <logicalName>,
    "database.history.kafka.bootstrap.servers":<>,
    "database.history.kafka.topic": "schema-changes.inventory",
    "plugin.name":"wal2json",
    "config.storage.replication.factor": "3",
    "offset.storage.replication.factor" : "3",
    "auto.create.topics.enable" : "true",
    "snapshot.mode" : "always"
  }
}

我在连接日志中看到的错误是，

2018-08-09 15:28:50,409 - DEBUG [KafkaBasedLog Work Thread - kconnect-offsets:Fetcher@199] - [Consumer clientId=consumer-1, groupId=1] Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(), toForget=(), implied=(kconnect-offsets-10, kconnect-offsets-4, kconnect-offsets-16, kconnect-offsets-7, kconnect-offsets-19, kconnect-offsets-13, kconnect-offsets-22, kconnect-offsets-1)) to broker kafka-02.hotel02.pro06.eu.idealo.com:9092 (id: 2002 rack: pro06)
2018-08-09 15:28:50,465 - DEBUG [kafka-producer-network-thread | producer-6:NetworkClient$DefaultMetadataUpdater@927] - [Producer clientId=producer-6] Sending metadata request (type=MetadataRequest, topics=dbserver1.public.spatial_ref_sys) to node kafka-01.hotel02.pro05.eu.idealo.com:9092 (id: 2004 rack: pro05)
2018-08-09 15:28:50,467 - WARN  [kafka-producer-network-thread | producer-6:NetworkClient$DefaultMetadataUpdater@882] - [Producer clientId=producer-6] Error while fetching metadata with correlation id 23856 : {dbserver1.public.spatial_ref_sys=UNKNOWN_TOPIC_OR_PARTITION}
2018-08-09 15:28:50,467 - DEBUG [kafka-producer-network-thread | producer-6:Metadata@270] - Updated cluster metadata version 23852 to Cluster(id = BwqlZApfT-ygzWr_wPcdng, nodes = [kafka-03.hotel02.pro05.eu.idealo.com:9092 (id: 2003 rack: pro05), kafka-01.hotel02.pro05.eu.idealo.com:9092 (id: 2004 rack: pro05), kafka-02.hotel02.pro06.eu.idealo.com:9092 (id: 2002 rack: pro06)], partitions = [])

【问题讨论】：

如果在属性文件中明确禁用，Kafka 不会自动创建主题
感谢@cricket_007，问题是，因为 aut0.create.topics 在 kafka 配置中被禁用，因此没有表名的主题被反映为 kafka 主题，并且当 kafka 被点击以获取主题的元数据时，它会引发 UNKNOWN _TOPIC_OR_PARTITION.通过启用它解决了它

标签： debezium postgresql apache-kafka apache-kafka-connect cdc debezium

【解决方案1】：

您在日志中看到的消息是警告，而不是错误。您能否尝试kafka-topics.sh 实用程序来列出可用主题？

【讨论】：