Kubernetes环境下Kafka Streams状态存储选型与配置咨询

1. Kafka Streams 状态存储是否与应用实例共置？ #

是的，Kafka Streams的状态存储默认与应用实例共置。在Kubernetes环境中，状态存储会部署在同一个Pod内：

• 如果使用内存存储，状态直接保存在应用进程的内存空间中，完全绑定当前Pod；
• 如果使用RocksDB，状态数据会存储在Pod的本地文件系统（默认路径为/tmp/kafka-streams），同样与Pod共置，不会脱离Pod独立存在。

2. Kubernetes中选择RocksDB还是内存存储更合适？ #

两者适用场景不同，需根据业务需求判断：

内存存储 #

• 适用场景：状态数据量小、Pod重启后状态可通过重新消费Kafka主题快速重建、对延迟要求极高的场景（比如简单的计数或过滤逻辑）。
• 缺点：Pod销毁或重启时状态会完全丢失；状态大小受Pod内存配额限制，无法存储大量数据。

RocksDB #

• 适用场景：状态数据量大、需要保留状态（Pod重启后无需重新全量消费）、内存资源有限的生产级场景。在K8s中，可结合emptyDir（临时持久化，Pod销毁后丢失）或PersistentVolume（持久化存储，Pod重建后状态保留）使用。
• 缺点：读写延迟略高于内存存储；需要配置合适的本地存储路径，避免因Pod调度导致状态丢失。

总结：Kubernetes生产环境中，大部分场景优先选择RocksDB；仅当状态量极小且无持久化需求时，才考虑内存存储。

3. 如何在应用中配置状态存储类型？ #

可通过全局配置或单个状态存储自定义配置两种方式设置：

全局配置 #

通过StreamsConfig设置默认的状态存储类型，应用中所有未单独指定的状态存储都会使用该配置：

• 设置全局默认内存存储：

Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_KEY_VALUE_STORE_SUPPLIER_CLASS_CONFIG, 
          "org.apache.kafka.streams.state.internals.InMemoryKeyValueStoreSupplier");

• 设置全局默认RocksDB存储（Kafka Streams默认即为RocksDB，可省略此配置，若需显式指定）：

Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_KEY_VALUE_STORE_SUPPLIER_CLASS_CONFIG, 
          "org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreSupplier");

单个状态存储配置 #

针对特定的状态存储单独指定类型，优先级高于全局配置：

内存存储示例

// 独立创建内存状态存储并添加到拓扑
StoreBuilder<KeyValueStore<String, Long>> inMemoryStore = Stores.keyValueStoreBuilder(
    Stores.inMemoryKeyValueStore("my-in-memory-store"),
    Serdes.String(),
    Serdes.Long()
);
streamsBuilder.addStateStore(inMemoryStore);

// 在聚合操作中指定内存存储
streamsBuilder.stream("input-topic")
    .groupByKey()
    .aggregate(() -> 0L,
        (key, value, agg) -> agg + value,
        Materialized.<String, Long>as("in-memory-agg-store")
            .withStoreSupplier(Stores.inMemoryKeyValueStore("in-memory-agg-store"))
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long())
    );

RocksDB存储示例

// 独立创建RocksDB状态存储并添加到拓扑
StoreBuilder<KeyValueStore<String, Long>> rocksDBStore = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("my-rocksdb-store"),
    Serdes.String(),
    Serdes.Long()
);
streamsBuilder.addStateStore(rocksDBStore);

// 在聚合操作中使用默认RocksDB存储（无需指定storeSupplier）
streamsBuilder.stream("input-topic")
    .groupByKey()
    .aggregate(() -> 0L,
        (key, value, agg) -> agg + value,
        Materialized.as("rocksdb-agg-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long())
    );

额外配置（RocksDB存储路径） #

对于RocksDB，可通过StreamsConfig.STATE_DIR_CONFIG自定义存储路径，比如挂载K8s的PersistentVolume路径：

Properties props = new Properties();
props.put(StreamsConfig.STATE_DIR_CONFIG, "/mnt/kafka-streams-state");

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