为什么我必须使用 Kafka Streams 配置状态存储

Question

目前我有以下设置：

StoreBuilder storeBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("kafka.topics.table"),
    new SomeKeySerde(),
    new SomeValueSerde());

streamsBuilder.addStateStore(storeBuilder);

final KStream<byte[], SomeClass> requestsStream = streamsBuilder
            .stream("myTopic", Consumed.with(Serdes.ByteArray(), theSerde));
    requestsStream
            .filter((key, request) -> Objects.nonNull(request))
            .process(() -> new SomeClassUpdater("kafka.topics.table", maxNumMatches), "kafka.topics.table");

Properties streamsConfiguration = loadConfiguration();
KafkaStreams streams = new KafkaStreams(streamsBuilder.build(), streamsConfiguration);

streams.start()

为什么我需要本地状态存储，因为我没有使用它进行任何其他计算，并且数据也存储在 kafka 更改日志中？还有它在什么时候存储在本地存储中，是否存储并提交到更改日志？

我面临的问题是我在本地存储，并且及时我遇到了内存问题，尤其是当它经常重新分区时。因为旧分区仍然闲置并填满内存。 所以我的问题是，为什么我们需要使用 RocksDB 的持久性，因为：

1. 数据保存在kafka changelog中
2. 当容器消失时，ramdisk 也消失了。

Answer 1

在单个线程上，我们可以有多个等于 no 的任务。主题的分区。每个分区都有自己的状态存储，这些状态存储将数据保存到 Kafka 内部主题的变更日志中。 分区的每个状态存储还维护其他分区的状态存储的副本，为了恢复任务可能失败的分区的数据。

如果您不使用状态存储，并且您的一项任务失败，它将转到内部主题，即 Changelog，然后将为分区获取数据，这对 CPU 来说是一项耗时的工作。因此，维护状态存储减少了任务可能失败的时间，并立即从另一个任务状态存储中获取数据。