如何为特定Kafka Topic配置消息留存？（Debezium CDC场景）

针对你团队基于Debezium实现CDC的Kafka集群，配置特定Topic的消息留存策略可从静态创建指定和运行时动态修改两种方式入手，以下是具体配置建议与操作步骤：

一、核心留存控制参数 #

Kafka通过两个核心参数控制消息留存，可根据业务需求组合使用：

• retention.ms：消息在Topic中留存的毫秒数，默认7天（604800000），设为-1表示永久留存。
• retention.bytes：Topic允许存储的最大总字节数，达到阈值后自动删除最早消息，默认-1（无限制）。

Debezium场景额外注意 #

Debezium会自动生成格式为serverName.databaseName.tableName的CDC Topic，若需修改这类自动创建Topic的策略：

• 可通过Kafka Broker的默认Topic配置批量预设，或对已存在Topic动态修改；
• 若希望Debezium创建Topic时直接应用自定义策略，需结合Kafka版本（2.6+）支持的Connector级配置实现。

二、具体操作方法 #

1. 静态配置：创建Topic时指定策略 #

若手动创建CDC相关Topic（不依赖Debezium自动创建），可在创建命令中直接指定参数：

kafka-topics.sh --create \
  --bootstrap-server your-kafka-broker:9092 \
  --topic your-debezium-cdc-topic \
  --partitions 3 \
  --replication-factor 2 \
  --config retention.ms=86400000 \  # 留存1天
  --config retention.bytes=1073741824  # 最大存储1GB

2. 动态配置：修改已存在的Topic #

针对Debezium自动创建的现有Topic，用kafka-configs.sh在线修改：

kafka-configs.sh --alter \
  --bootstrap-server your-kafka-broker:9092 \
  --topic your-debezium-cdc-topic \
  --add-config retention.ms=604800000 \  # 恢复默认7天留存
  --add-config retention.bytes=-1  # 取消存储大小限制

如需批量修改匹配前缀的CDC Topic，可结合脚本实现：

# 批量修改所有以debezium-server开头的Topic留存为2天
kafka-topics.sh --list --bootstrap-server your-kafka-broker:9092 | grep "^debezium-server" | while read topic; do
  kafka-configs.sh --alter \
    --bootstrap-server your-kafka-broker:9092 \
    --topic $topic \
    --add-config retention.ms=172800000
done

三、验证配置生效 #

修改完成后，用以下命令查看Topic当前配置：

kafka-configs.sh --describe \
  --bootstrap-server your-kafka-broker:9092 \
  --topic your-debezium-cdc-topic

输出中会显示retention.ms和retention.bytes的当前值，确认是否符合预期。

四、Debezium Connector级预设策略（Kafka 2.6+） #

若希望Debezium自动创建Topic时直接应用自定义留存策略，可在Connector配置文件中添加：

# 为所有自动创建的Topic设置默认留存规则
topic.creation.default.retention.ms=86400000
topic.creation.default.retention.bytes=1073741824

# 针对特定前缀的Topic设置专属规则
topic.creation.patterns.debezium-cdc.prefix=my-server-.*
topic.creation.patterns.debezium-cdc.retention.ms=172800000

需确保Connector配置中topic.creation.enable设为true（默认启用）。

内容的提问来源于stack exchange，提问作者Adhi Putra