Kafka Connect和Key Converter的主题名称策略

Question

我正在尝试设置Debezium MySQL源连接器。我的目标是为每个数据库提供一个主题，因此我正在研究利用主题的可能性，以使主题可以包含不同的消息类型并且它们的模式可以存储在ConFluent Schema注册表中。

根据这里的几个答案，我将键和值转换器主题名策略设置为io.confluent.kafka.serializers.subject.TopicRecordNameStrategy。

要将来自同一模式的所有消息重新路由到同一主题，我使用以下配置：

{
  "name": "aws-db-connector",
  "config": {
    "group.id": "aws-db-group",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "secret-pw",
    "database.server.id": "184054",
    "database.server.name": "aws-db",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.aws-db",
    "database.include.list": "db1,db2",
    "transforms": "unwrap,Reroute",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.unwrap.add.fields": "db,table,op,source.ts_ms",
    "transforms.Reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
    "transforms.Reroute.topic.regex": "(.*\\S)\\.(.*\\S)\\.(.*\\S)",
    "transforms.Reroute.topic.replacement": "$2_schema",
    "transforms.Reroute.key.field.name": "table",
    "transforms.Reroute.key.field.regex": "(.*\\S)\\.(.*\\S)\\.(.*\\S)",
    "transforms.Reroute.key.field.replacement": "$3"
  }
}

在我的docker-compose文件中，我设置了：

- CONNECT_KEY_CONVERTER_KEY_SUBJECT_NAME_STRATEGY=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
- CONNECT_VALUE_CONVERTER_VALUE_SUBJECT_NAME_STRATEGY=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
- CONNECT_KEY_CONVERTER=io.confluent.connect.avro.AvroConverter
- CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL=http://registry:8081
- CONNECT_VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter
- CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL=http://registry:8081

对于值，这是完美的工作。我可以看到我的模式注册表包含多个主题，格式为

不幸的是，对于键，这个策略没有按预期工作，我只有一个模式主题：它看起来像是包含新主题名而不是原始主题名。如果一个字段名在不同的表中具有不同的类型，则会导致冲突和不兼容错误。

生成密钥主题时，是否有任何方法可以提供适当的RecordName？

编辑-添加示例：

假设我的数据库包含三个表，table1、table2和table3。

表1：

CREATE TABLE `table1` (
    `id` INT NOT NULL AUTO_INCREMENT,
    `name` TEXT,
    PRIMARY KEY (`id`)
);

表2：

CREATE TABLE `table2` (
    `id` INT NOT NULL AUTO_INCREMENT,
    `name` BINARY,
    PRIMARY KEY (`id`)
);

表3：

CREATE TABLE `table3` (
    `id` BINARY NOT NULL,
    `name` INT,
    PRIMARY KEY (`id`)
);

使用上述配置运行Debezium，它会在模式注册表中创建以下值主题：

• db1_schema.db1-aws-db。表1数值

以及以下关键主题：

• db1\u schema.db1\u schema-Key

轮到table3时，Debezium连接器失败，因为id列以int类型注册在模式注册表主题中，并且它与table3中的bytes类型不兼容。因此我得到了这个错误：

正在注册的模式与早期模式不兼容；错误代码：409

我所期望的是，也会为密钥创建单独的主题：

• db1_模式。aws-db-db1。表1键
• db1_模式。aws-db-db1。表2键
• db1_schema.aws-db-db1.table3-Key

这样，具有不同键模式的消息可以存储在同一个主题中。




Answer 1

这似乎是默认情况下Debezium的工作方式。它只会为每个主题创建一个键模式，但会创建不同的值模式，因此重新路由到该主题的所有消息都应该共享相同的键结构。

为了解决这个问题，应该使用RegexRouter。在重新路由之前应用InsertField转换也会将原始主题名添加到键中，并且可以从中提取表名。

"transforms": "InsertField,Reroute",
"transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Key",
"transforms.InsertField.topic.field": "table"
"transforms.Reroute.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.Reroute.regex": "(.*\\S)\\.(.*\\S)\\.(.*\\S)",
"transforms.Reroute.replacement": "$2_schema",