如何以避免重新创建状态存储的方式重新启动 KafkaStreams 拓扑

Question

我用KafkaStreams DSL和处理器api编写了一些代码来实时计算：

  KStreamBuilder builder = new KStreamBuilder();
  KStream<String, String> logs = builder.stream(stringSerde, stringSerde, ADDCASH_TOPIC_NAME);
  StateStoreSupplier countStore = Stores.create("Counts")
            .withKeys(Serdes.String())
            .withValues(Serdes.Long())
            .persistent()
            .build();
 logs.process(() -> new AddCashProcessor(), countStore.name());

我想知道：当我重新启动拓扑时，Stores.create(..)方法是否会重新创建状态存储？

Answer 1

它将重用现有状态，并且仅在没有时创建和清空/新建存储。