能否配置Kafka不删除未读消息？合规场景特殊存储规则咨询

好问题！先直接给你结论：Kafka本身没有原生支持“基于存储容量限制，优先保留未消费消息并在达到限制时抛出生产者异常”的配置，但我们可以通过组合Kafka的现有特性+外部监控/自动化脚本，来实现接近你需求的效果。

先理清Kafka的默认行为 #

首先得明确几个关键点：

• 你提到的log.retention.bytes（对应你说的25GB）是全局或话题级的日志容量限制，默认规则是不管消息是否被消费，只要总容量达到阈值，就会删除最早的日志段，不会考虑未消费的消息。
• 磁盘满时生产者抛出异常，是Kafka的底层磁盘保护机制（由disk.full.policy控制，新版本默认会拒绝写入），这是磁盘空间不足触发的，和日志保留策略是两个独立的逻辑。

实现需求的可行方案 #

要满足“合规话题消息不被删除（只要未被消费）+ 达到25GB时阻止生产者写入”，可以按以下步骤操作：

1. 给合规话题配置基础保留规则 #

首先针对合规话题单独设置保留策略，避免全局配置影响其他话题：

• 设置话题级的retention.bytes=30GB（比目标25GB稍大，留缓冲空间）
• 设置retention.ms=-1（禁用基于时间的自动删除，确保只有容量相关的规则生效）
• 保持默认的log.cleanup.policy=delete（因为我们不需要日志压缩）

你可以用Kafka命令行工具创建话题时指定这些参数：

kafka-topics.sh --create --topic compliance-topic --bootstrap-server your-kafka-broker:9092 \
  --partitions 3 --replication-factor 2 \
  --config retention.bytes=26843545600 \  # 30GB，按字节计算
  --config retention.ms=-1

2. 监控话题容量与未消费消息状态 #

接下来需要监控两个核心指标：

• 合规话题的实际磁盘占用量（可以通过Kafka的kafka-log-dirs.sh工具，或Prometheus抓取kafka_log_log_size指标）
• 消费者组的延迟（即log-end-offset - current-offset，用来判断是否有未被消费的消息）

3. 触发生产者写入阻止逻辑 #

当监控到以下两个条件同时满足时，自动触发阻止生产者写入的操作：

1. 合规话题的磁盘占用量接近或达到25GB
2. 存在未被消费的消息（消费者延迟>0）

具体阻止写入的方式有两种：

方式一：动态调整话题的ISR配置

通过修改话题的min.insync.replicas为大于当前可用副本数的值，这样生产者写入时会因为无法满足同步副本要求而抛出NotEnoughReplicas异常：

# 假设话题有2个副本，将min.insync.replicas设为3（大于可用副本数）
kafka-configs.sh --alter --topic compliance-topic --bootstrap-server your-kafka-broker:9092 \
  --add-config min.insync.replicas=3

当未消费消息处理完毕、话题占用量下降后，再将配置改回原值（比如1或2）。

方式二：使用ACL临时禁止生产者写入

如果你的Kafka启用了ACL，可以临时移除生产者的写入权限，这样生产者会收到AuthorizationException异常：

# 移除某个生产者用户的写入权限
kafka-acls.sh --remove --topic compliance-topic --bootstrap-server your-kafka-broker:9092 \
  --allow-principal User:producer-user --operation Write

同样，在条件解除后恢复权限。

注意事项 #

• 自动化脚本是关键：以上监控和触发操作需要通过脚本（比如Python、Shell）或监控工具（比如Prometheus Alertmanager + 自定义Webhook）来自动执行，否则人工操作会有延迟。
• 缓冲空间的设置：retention.bytes设为比25GB大，是为了避免在未消费消息存在时，Kafka自动删除旧消息，给我们留出足够的时间触发阻止逻辑。
• 测试验证：一定要在测试环境中验证整个流程，确保触发条件和阻止逻辑符合预期，避免影响生产环境。

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