Kafka Streams：能否为状态存储设置‘compact,delete’策略及保留规则？

当然可以！Kafka Streams的状态存储底层依赖对应的changelog Topic来持久化状态变更，所以你完全可以通过配置changelog Topic的参数，实现compact,delete混合清理策略，同时搭配自定义保留规则。下面我详细拆解具体操作和注意事项：

Kafka Streams的持久化状态存储（比如KeyValueStore）会自动生成对应的changelog Topic，所有状态变更都会写入这个Topic。状态存储的清理策略本质上就是changelog Topic的清理策略，因此我们只需要给目标changelog Topic配置compact,delete策略及相关保留规则即可。

你有两种配置维度可选，按需选择即可：

1. 全局配置（所有状态存储生效） #

在初始化StreamsConfig时，设置全局的changelog参数，所有状态存储的changelog Topic都会继承这个配置：

Properties props = new Properties();
// 基础配置（示例）
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

// 全局配置changelog的混合清理策略与保留规则
props.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
props.put(TopicConfig.RETENTION_MS_CONFIG, "604800000"); // 保留7天
props.put(TopicConfig.DELETE_RETENTION_MS_CONFIG, "86400000"); // 删除标记保留24小时
props.put(TopicConfig.RETENTION_BYTES_CONFIG, "5368709120"); // 最大占用5GB存储空间

KafkaStreams streams = new KafkaStreams(topology, props);

2. 单个状态存储配置（精细化控制） #

如果只想给特定状态存储配置该策略，可以在创建存储时通过StoreBuilder的withLoggingEnabled()方法单独指定：

// 创建自定义配置的状态存储
StoreBuilder<KeyValueStore<String, UserProfile>> userStoreBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("user-profile-store"),
    Serdes.String(),
    new UserProfileSerde()
)
.withLoggingEnabled(Map.of(
    TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete",
    TopicConfig.RETENTION_MS_CONFIG, "86400000", // 保留1天
    TopicConfig.DELETE_RETENTION_MS_CONFIG, "3600000", // 删除标记保留1小时
    TopicConfig.RETENTION_BYTES_CONFIG, "1073741824" // 最大占用1GB
));

// 将状态存储添加到拓扑中
topology.addStateStore(userStoreBuilder);

• cleanup.policy=compact,delete：混合策略，Kafka会同时执行两种逻辑：一是按compact策略保留每个键的最新版本；二是按delete策略删除超过保留时间/存储空间限制的所有数据。
• retention.ms：数据的最大保留时长，超过该时间的所有数据（无论是否为键的最新值）都会被删除。
• retention.bytes：Topic允许占用的最大存储空间，超出后会从最旧的数据开始删除。
• delete.retention.ms：压缩策略中，删除标记（tombstone，用于标记某个键被删除）的保留时间。设置过短可能导致压缩未完成就丢失标记，引发状态恢复不一致；过长则会浪费存储空间，默认24小时，可根据业务场景调整。

• 集群版本要求：确保你的Kafka集群版本在0.11.0.0及以上，这个版本开始正式支持compact,delete混合清理策略。
• 策略优先级：压缩和删除逻辑是并行执行的，Kafka后台线程会同时处理压缩任务和过期/超量数据删除，既保证状态的最新性，又能及时清理冗余数据。
• 状态恢复风险：配置保留规则时要留足冗余，确保changelog Topic中有足够数据支撑状态恢复，避免因数据被提前删除导致状态恢复不完整。

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