apache-kafka - Debezium MySQL CDC 连接器：org.apache.kafka.connect.errors.ConnectException：遇到表的更改事件

Question

Debezium 连接器版本：0.9.2.Final

数据库历史主题：testing_engagex_dbhistory_test

保留时间：-1（无限制）

保留字节数：-1（无限制）

连接器配置

{
  name: 'Mysql-cdc-engagex-test',
  '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',
  'database.history.producer.ssl.truststore.location': '/path/to/keys/public.truststore.jks',
  'value.converter.schemas.enable': 'false',
  'database.history.consumer.ssl.truststore.location': '/path/to/keys/public.truststore.jks',
  'database.history.producer.ssl.truststore.password': 'password',
  'database.history.producer.ssl.keystore.location': '/path/to/keys/public.keystore.p12',
  'database.history.consumer.ssl.truststore.password': 'password',
  'database.history.consumer.ssl.keystore.location': '/path/to/keys/public.keystore.p12',
  'database.history.producer.ssl.keystore.type': 'PKCS12',
  'database.history.producer.ssl.keystore.password': '******',
  'database.history.consumer.ssl.key.password': ''******',
  'database.history.producer.security.protocol': 'SSL',
  'database.history.consumer.ssl.keystore.type': 'PKCS12',
  'database.history.consumer.ssl.keystore.password': ''******',
  'database.history.producer.ssl.key.password': ''******',
  'database.history.consumer.security.protocol': 'SSL',
  'key.converter.schemas.enable': 'false',
  'errors.log.enable': 'true',
  'errors.log.include.messages': 'true',
  'database.hostname': '<hostname>',
  'database.port': '3306',
  'database.user': '<username>',
  'database.password': '<password>',
  'database.server.name': 'testing_engagex_cdc_test',
  'database.history.kafka.bootstrap.servers': '<serverUrl>',
  'database.history.kafka.topic': 'testing_engagex_dbhistory_test',
  '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': '<db1,db2,db3>',
  'table.whitelist':'<whitelist tables comma separated>',
  'column.blacklist':'<blacklist column comma separated>',
  'gtid.source.filter.dml.events': 'true',
  'tombstones.on.delete': 'true',
  'connect.keep.alive': 'true',
  'snapshot.minimal.locks': 'true'
}

我的Debezium MySQL CDC 连接器运行良好。突然它因以下错误而停止。提到的表“i9.workshop”中没有架构更改。我不明白为什么会出现这个错误？

我删除了“i9.workshop”并在另一个表中遇到了同样的问题。（对任何表都没有架构更改）

org.apache.kafka.connect.errors.ConnectException: Encountered change event for table i9.workshopwhose 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 i9.workshopwhose 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)

正如来自不同博客的一些人所建议的那样，我将 snapshot.mode 更改为 schema_only 以进行恢复并再次收到以下错误。

org.apache.kafka.connect.errors.ConnectException: 35
    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: java.lang.ArrayIndexOutOfBoundsException: 35
    at io.debezium.relational.TableSchemaBuilder.lambda$createValueGenerator$2(TableSchemaBuilder.java:210)
    at io.debezium.relational.TableSchema.valueFromColumnData(TableSchema.java:135)
    at io.debezium.connector.mysql.RecordMakers$1.insert(RecordMakers.java:236)
    at io.debezium.connector.mysql.RecordMakers$RecordsForTable.create(RecordMakers.java:443)
    at io.debezium.connector.mysql.BinlogReader.handleInsert(BinlogReader.java:792)
    at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:461)
    ... 5 more

您能否向我解释为什么会发生此错误以及我们如何永久解决它？

标签： apache-kafkadebezium