Debezium同步之Kafka Connect 自动创建主题的自定义

topic.creation.enable = true

一组主题属性与主题创建组相关联。最低限度，您必须定义一个default主题创建组并指定其配置属性。除此之外，您还可以选择定义一个或多个自定义主题创建组并为每个组指定唯一属性。

创建自定义主题创建组时，您可以根据主题名称模式为每个组定义成员主题。您可以指定命名模式来描述要从每个组中包含或排除的主题。include和exclude属性包含定义主题名称模式的正则表达式的逗号分隔列表。例如，如果您希望组包含以字符串 开头的所有主题，请将其属性dbserver1.inventory的值设置为。topic.creation.inventory.includedbserver1\\.inventory\\.*

主题创建组和每个default自定义组都与一组唯一的配置属性相关联。您可以配置组以包含任何Kafka 主题级配置属性。例如，您可以指定旧主题段的清理策略、保留时间或主题组的主题压缩类型。您必须至少定义一组属性来描述要创建的主题的配置。

如果没有注册自定义组，或者include任何注册组的模式与要创建的任何主题的名称不匹配，则 Kafka Connect 使用default组的配置来创建主题。

有关通用主题配置注意事项，请参阅Debezium 安装指南中的配置 Debezium 主题。

在您可以使用 Kafka Connect 自动创建主题之前，您必须创建一个默认主题创建组并为其定义配置。默认主题创建组的配置应用于名称与include自定义主题创建组的列表模式不匹配的任何主题。

• 要为topic.creation.default组定义属性，请将它们添加到连接器配置 JSON，如以下示例所示：

  {
      ...
  
      "topic.creation.default.replication.factor": 3,  
      "topic.creation.default.partitions": 10,  
      "topic.creation.default.cleanup.policy": "compact",  
      "topic.creation.default.compression.type": "lz4"  
  
       ...
  }

  您可以在组的配置中包含任何Kafka 主题级别的配置属性default。

您可以定义多个自定义主题组，每个主题组都有自己的配置。

• 要定义自定义主题组，请将属性添加到连接器 JSON，并在组名称后列出自定义组的属性。topic.creation.<group_name>.include

  以下示例显示了自定义主题创建组inventory的示例配置：applicationlogs

  {
      ...
  
      
      "topic.creation.inventory.include": "dbserver1\\.inventory\\.*",  
      "topic.creation.inventory.partitions": 20,
      "topic.creation.inventory.cleanup.policy": "compact",
      "topic.creation.inventory.delete.retention.ms": 7776000000,
  
      
      "topic.creation.applicationlogs.include": "dbserver1\\.logs\\.applog-.*",  
      "topic.creation.applicationlogs.exclude": "dbserver1\\.logs\\.applog-old-.*",  
      "topic.creation.applicationlogs.replication.factor": 1,
      "topic.creation.applicationlogs.partitions": 20,
      "topic.creation.applicationlogs.cleanup.policy": "delete",
      "topic.creation.applicationlogs.retention.ms": 7776000000,
      "topic.creation.applicationlogs.compression.type": "lz4",
  
       ...
  }

为任何自定义主题创建组指定配置后，注册组。

• 通过将topic.creation.groups属性添加到连接器 JSON 并指定以逗号分隔的组列表来注册自定义组。

  以下示例注册自定义主题创建组inventory和applicationlogs：

  {
      ...
  
      "topic.creation.groups": "inventory,applicationlogs",
  
      ...
  }

以下示例显示了一个完整的配置，其中包括default主题组的配置，inventory以及applicationlogs自定义主题创建组的配置：