如何为Kafka集群全局配置消息1小时默认保留策略？

你可以通过修改Broker全局配置和批量更新已存在Topic配置两步来实现，无需逐个手动配置：

1. 设置Broker全局默认规则（覆盖新创建的Topic） #

修改Kafka集群每个Broker节点的server.properties配置文件，添加或调整以下参数：

• log.retention.hours=1：全局默认消息保留时长为1小时
• log.retention.check.interval.ms=300000（可选，为默认值）：控制Kafka检查过期消息的频率（5分钟一次），确保过期消息能及时被清理

修改完成后重启所有Broker节点。之后新创建的所有Topic都会自动继承这个1小时的保留策略，无需手动指定配置。

2. 批量更新已存在的数千个Topic #

针对已创建好的Topic，用Shell脚本批量修改它们的消息保留配置：

方法一：循环遍历批量修改 #

# 替换为你的Kafka集群bootstrap地址（例如xxx.xxx.xxx.xxx:9092）
BOOTSTRAP_SERVER="localhost:9092"

# 遍历所有Topic，设置retention.ms为3600000毫秒（即1小时）
kafka-topics.sh --list --bootstrap-server $BOOTSTRAP_SERVER | while read topic; do
  kafka-configs.sh --alter --bootstrap-server $BOOTSTRAP_SERVER --entity-type topics --entity-name $topic --add-config retention.ms=3600000
done

方法二：用xargs提升执行效率 #

kafka-topics.sh --list --bootstrap-server localhost:9092 | xargs -I {} kafka-configs.sh --alter --bootstrap-server localhost:9092 --entity-type topics --entity-name {} --add-config retention.ms=3600000

验证配置生效 #

执行以下命令查看单个Topic的配置，确认retention.ms是否为3600000：

kafka-configs.sh --describe --bootstrap-server <你的Kafka地址> --entity-type topics --entity-name <目标Topic名称>

注意：如果部分Topic之前设置过自定义保留时长，上述批量脚本会直接覆盖原有配置，实现全局统一的1小时保留规则。

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