MySQL Debezium Kafka：此连接器不知道架构

Question

我用 "table.whitelist" 中的一个表启动了 MySQL Debezium Kafka 连接器（版本：0.9.2.Final），它工作正常。在白名单中添加另一个表并重新启动连接器时，我收到以下错误。

org.apache.kafka.connect.errors.ConnectException: Encountered change event for table paperclip.iltwhose schema isn't known to this connector
    at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:230)
    at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:208)
    at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:477)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1095)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:943)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:580)
    at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:825)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.connect.errors.ConnectException: Encountered change event for table paperclip.iltwhose schema isn't known to this connector
    at io.debezium.connector.mysql.BinlogReader.informAboutUnknownTableIfRequired(BinlogReader.java:727)
    at io.debezium.connector.mysql.BinlogReader.handleUpdateTableMetadata(BinlogReader.java:702)
    at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:461)
    ... 5 more

请找到我使用的以下配置。我希望通过这个设置（"database.history.store.only.monitored.tables.ddl": "false"），它应该可以工作。

我该如何解决这个案子？

{
    "name": "Mysql-rnd-engagex",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "3",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "database.hostname": "devmysql.xxxx.net",
    "database.port": "3306",
    "database.user": "xxxxxx",
    "database.password": "xxxxx",
    "database.server.name": "rnd_engagex_cdc",
    "database.history.kafka.bootstrap.servers": "xxxxxx.aivencloud.com:xxxx",
    "database.history.kafka.topic": "rnd_engagex_dbhistory",
    "database.history.skip.unparseable.ddl": "false",
    "database.history.store.only.monitored.tables.ddl": "false",
    "include.schema.changes": "false",
    "include.query": "false",
    "table.ignore.builtin": "true",
    "database.whitelist": "paperclip",
    "table.whitelist": "paperclip.elearning",    //added new table : "paperclip.elearning,paperclip.ilt"
    "column.blacklist": "paperclip.elearning.description",
    "gtid.source.filter.dml.events": "true",
    "tombstones.on.delete": "true",
    "connect.keep.alive": "true",
    "snapshot.minimal.locks": "true",
    "database.history.producer.ssl.truststore.location": "/xxxx/yyyy/keys/public.truststore.jks",
    "value.converter.schemas.enable": "false",
    "database.history.consumer.ssl.truststore.location": "/xxxx/yyyy/keys/public.truststore.jks",
    "database.history.producer.ssl.truststore.password": "password",
    "database.history.producer.ssl.keystore.location": "/xxxx/yyyy/keys/public.keystore.p12",
    "database.history.consumer.ssl.truststore.password": "password",
    "database.history.consumer.ssl.keystore.location": "/xxxx/yyyy/keys/public.keystore.p12",
    "database.history.producer.ssl.keystore.type": "PKCS12",
    "database.history.producer.ssl.keystore.password": "ppppppppp",
    "database.history.consumer.ssl.key.password": "ppppppppp",
    "database.history.producer.security.protocol": "SSL",
    "database.history.consumer.ssl.keystore.type": "PKCS12",
    "database.history.consumer.ssl.keystore.password": "ppppppppp",
    "database.history.producer.ssl.key.password": "ppppppppp",
    "database.history.consumer.security.protocol": "SSL",
    "key.converter.schemas.enable": "false"
}

Answer 1

您需要在创建连接器时添加属性“snapshot.new.tables”：“parallel”，然后您才能在后期将更多表列入白名单。这在文档中没有给出，因为该功能在 0.9.x 中是 beta 版

Answer 2

就我而言，我已经设置了

"snapshot.mode" : "initial",

这就是解决方案。