Debezium 错误，此连接器不知道架构

Question

我有一个使用 Debezium 的项目，主要基于此示例，然后将其连接到 Apache Pulsar。

\n\n

我改变了一些配置。该文件现在如下所示：

\n\n

database.history=io.debezium.relational.history.MemoryDatabaseHistory\nconnector.class=io.debezium.connector.mysql.MySqlConnector\noffset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore\noffset.storage.file.filename=offset.dat\noffset.flush.interval.ms=5000\nname=mysql-dbz-connector\ndatabase.hostname={ip}\ndatabase.port=3308\ndatabase.user={user}\ndatabase.password={pass}\ndatabase.dbname=database\ndatabase.server.name=test\ntable.whitelist=database.history_table,database.project_table\nsnapshot.mode=schema_only\nschemas.enable=false        \ninclude.schema.changes=false\npulsar.topic=persistent://public/default/{0}\npulsar.broker.address=pulsar://{ip}:6650\ndatabase.history=io.debezium.relational.history.MemoryDatabaseHistory\n

\n\n

正如您可能理解的，我想要做的是监视数据库的修改history_table和修改，然后将有效负载写入 Apache Pulsar。project_table

\n\n

我的问题如下。无论我使用什么快照模式，当写入偏移量时，我都无法重新启动 Debezium，而不会在下一次数据库更新时出现错误。

\n\n

Encountered change event for table database.history_table whose schema isn\'t known to this connector\n

\n\n

它仅发生在现有offset.dat文件上。我认为这是因为文件中的架构为空offset.dat。以这个为例：

\n\n

\xc2\xa8\xc3\x8csrjava.util.HashMap\xe2\x81\x84\xc2\xa1\xe2\x88\x9a`\xe2\x80\x94F\nloadFactorI thresholdxp?@wur[B\xc2\xa8\xc3\x9b\xc2\xafT\xe2\x80\xa1xpG{"schema":null,"payload":["mysql-dbz-connector",{"server":"test"}]}uq~U{"ts_sec":1563802215,"file":"database-bin.000005","pos":79574,"server_id":1,"event":1}x\n

\n\n

我首先怀疑我用来使 JSON 更简洁的schemas.enable=false或include.schema.changes=false参数，但它们的值不会改变文件中的任何内容offset.dat。

\n

Answer 1

问题出在线路上database.history=io.debezium.relational.history.MemoryDatabaseHistory。历史记录将无法重新启动。您应该使用 FileDatabaseHistory 而不是它。