Kafka 主题级配置从创建到优化

一、为什么要关心“主题级配置”？

Kafka 的很多行为由 Broker 默认值 决定，但同一集群里不同业务的主题往往诉求各异（如交易流与埋点流）。通过主题级覆盖值你可以精细化地对单个 Topic 做容量、保留、压缩、耐久和复制限速等调整，而不影响其他主题。

二、创建/修改/查询/删除覆盖值

1）创建时设置覆盖项

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic my-topic --partitions 1 --replication-factor 1 \
  --config max.message.bytes=64000 --config flush.messages=1

2）运行期修改（可叠加）

bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name my-topic \
  --alter --add-config max.message.bytes=128000

3）查看当前覆盖项

bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name my-topic --describe

4）删除覆盖项（恢复为 Broker 默认或静态值）

bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name my-topic \
  --alter --delete-config max.message.bytes

三、核心维度与最佳实践

1）保留策略与压缩：容量与可追溯性的平衡

• cleanup.policy：delete（默认）按时间/容量清理；compact 仅保留每个 key 的最新值；也可 delete,compact 组合。
• retention.ms/bytes：按时间/容量控制“删除”策略的阈值。注意它对分区级生效：主题实际容量 ≈ retention.bytes × 分区数。
• local.retention.ms/bytes：启用分层存储（见第 6 点）时，控制本地段的保留上限。
• delete.retention.ms：压缩主题里“删除墓碑（tombstone）”的保留窗口，决定了消费者从 0 读到最终一致快照所需在该窗口内完成。
• min.cleanable.dirty.ratio、min/max.compaction.lag.ms：控制何时触发压缩以及记录在未压缩状态下保留的时间。
• segment.ms/bytes、segment.jitter.ms：段滚动的时间与大小颗粒。更小的段提升清理/压缩的颗粒度，但文件更多。

建议：

• 交易/状态类主题 → compact 或 delete,compact；配合合理的 delete.retention.ms 保障补扫窗口。
• 纯日志/埋点类主题 → delete + 明确 retention.ms/bytes，别只依赖时间或只依赖容量。
• 大并发场景下给 segment.jitter.ms 一个非零抖动，避免“齐步滚段”引发抖动风暴。

2）压缩算法与吞吐：延迟与成本的折中

• compression.type：producer（保留生产者侧设置，默认）、zstd、lz4、snappy、gzip、uncompressed。
• compression.*.level：针对 gzip / lz4 / zstd 的级别细调。

建议：

• 优先考虑 zstd（默认 level=3）作为高压缩比的通用解；对极致低延迟可选 lz4。
• 端到端一致性：大量接入方时，统一 compression.type 便于观测与故障排查。

3）大消息与批量

• max.message.bytes：限制批的大小（新格式总是批）。若增大，请同步提升老版本消费者（< 0.10.2）的 fetch 大小。
• flush.messages / flush.ms：强制 fsync 的条数/时间间隔。不推荐依赖它确保持久化，交给复制与 OS 刷新更高效。

建议：

• 谨慎放大 max.message.bytes；更提倡应用层切分与批压缩。
• 避免开启频繁 fsync；生产环境一般不设置 flush.*。

4）时间戳语义与有效性校验

• message.timestamp.type：CreateTime（消息创建时间，默认）或 LogAppendTime（Broker 追加时间）。
• message.timestamp.after.max.ms / before.max.ms：限制消息时间戳与 Broker 时间戳的最大正/负偏移，超限将拒收（仅在 CreateTime 下生效）。

建议：

• 事件时间驱动的计算保持 CreateTime；若遇到上游时钟漂移，合理放宽 before/after.max.ms 或修正上游时钟。

5）耐久与一致性：副本设置

• min.insync.replicas + 生产者 acks=all：构建强一致写入。常见组合：RF=3、min.insync.replicas=2、acks=all。
• unclean.leader.election.enable：允许非 ISR 选主（可能丢数）。仅在应急场景打开，KRaft 下动态启用生效可能有延迟（默认 5 分钟，可用 kafka-leader-election.sh --unclean 立即触发）。

建议：

• 对关键主题默认关闭不干净选主；容量/延迟不敏感但极度重一致的主题提升 min.insync.replicas。

6）分层存储（Tiered Storage）相关

• remote.storage.enable：为主题启用分层存储（当前版本启用后不可关闭）。
• remote.log.copy.disable：将分层数据置为只读并停止继续上传；此时 local.retention.* 不再生效，过期以 retention.* 为准。
• remote.log.delete.on.disable：准备关闭分层时，是否同时清理分层数据。

建议：

• 在开启前评估回退策略与成本/延迟曲线；结合 local.retention.* 做冷热数据分层。

7）复制节流与迁移

• leader.replication.throttled.replicas / follower.replication.throttled.replicas：在数据搬迁、跨机房复制或修复期间对特定副本限速，避免影响在线流量。

8）索引与查找

• index.interval.bytes：索引粒度；segment.index.bytes：索引文件大小。通常按默认值即可。

四、按目标调参的小抄

目标一：降低存储成本

• cleanup.policy=delete，合理 retention.ms/bytes
• 启用 zstd 压缩；适度增大 segment.bytes 减少元数据碎片
• 分层存储：remote.storage.enable=true + 合理 local.retention.*

目标二：降低端到端延迟

• lz4 或保留 producer，控制生产端批大小
• 避免频繁 fsync；更依赖复制与 OS 刷新
• 小心放宽时间戳校验以减少拒收重试

目标三：提升吞吐

• 生产端批量与压缩协同；主题侧保持 compression.type=producer
• 合理增大 segment.bytes，减少滚段频率
• 必要时使用复制节流避免后台复制冲击前台

目标四：提升可恢复性（强一致）

• RF=3、min.insync.replicas=2、生产端 acks=all
• 关闭 unclean.leader.election.enable
• 压缩主题设置足够大的 delete.retention.ms 保障补扫窗口

五、常见坑位与规避

• 容量误判：retention.bytes 作用于分区，别忘了乘以分区数。
• 意外滚段：retention.bytes=0 或 retention.ms 到点会触发滚段，留意对下游的影响。
• 版本不匹配：放大 max.message.bytes 时，老消费者（< 0.10.2）也要同步增大 fetch。
• 墓碑过早清理：delete.retention.ms 太小，消费者“全量+增量”补扫会读不到删除标记。
• KRaft 下不干净选主：动态开启有生效周期，必要时用工具立即触发。

六、为“订单状态流”定制一个高一致 + 可追溯主题

# 创建：压缩 + 删除并行，强一致 + 7 天追溯
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic order-status \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=delete,compact \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2 \
  --config compression.type=zstd \
  --config delete.retention.ms=172800000 \
  --config segment.ms=86400000 \
  --config message.timestamp.type=CreateTime

后续如需迁移/修复限速：

bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name order-status \
  --alter --add-config leader.replication.throttled.replicas=*:*