如何在两个Kafka流之间使用持久化状态存储

Question

我在试图通过Kafka流实现以下目标时遇到了一些困难：

• 在应用程序启动时，（压缩的）主题α被加载到键值StateStore中
• Kafka流从另一个主题中消费，使用上面的映射（get），并最终在主题alpha中生成一个新记录
• 结果是，即使拖缆重新启动，内存中的映射也应与底层主题对齐

我的方法如下：

val builder = new StreamsBuilderS()

val store = Stores.keyValueStoreBuilder(
   Stores.persistentKeyValueStore("store"), kSerde, vSerde)
)

builder.addStateStore(store)

val loaderStreamer = new LoaderStreamer(store).startStream()

[...] // I wait a few seconds until the loading is complete and the stream os running

val map = instance.store("store", QueryableStoreTypes.keyValueStore[K, V]()) // !!!!!!!! ERROR HERE !!!!!!!!

builder
  .stream("another-topic")(Consumed.`with`(kSerde, vSerde))
  .doMyAggregationsAndgetFromTheMapAbove
  .transform(() => new StoreTransformer[K, V]("store"), "store")
  .to("alpha")(Produced.`with`(kSerde, vSerde))

装载机Treamer（store）：

[...]
val builders = new StreamsBuilderS()

builder.addStateStore(store)

builder
  .table("alpha")(Consumed.`with`(kSerde, vSerde))

builder.build
[...]

存储变压器：

[...]
override def init(context: ProcessorContext): Unit = {
  this.context = context
  this.store = 
    context.getStateStore(store).asInstanceOf[KeyValueStore[K, V]]
}

override def transform(key: K, value: V): (K, V) = {
  store.put(key, value)
  (key, value)
}
[...]

...但是我得到的是：

Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException:
The state store, store, may have migrated to another instance.

试图获取存储处理程序时。

你知道如何做到这一点吗？

非常感谢。

Answer 1

您不能在两个Kafka Streams应用程序之间共享状态存储。

根据文件：https://docs.confluent.io/current/streams/faq.html#interactive-查询上述异常可能有两个原因：

>

本地KafkaStreams实例尚未准备好，因此还无法查询其本地状态存储。

本地KafkaStreams实例已准备就绪，但特定的状态存储刚刚迁移到幕后的另一个实例。

最简单的处理方法是等待状态存储可查询：

public static <T> T waitUntilStoreIsQueryable(final String storeName,
                                              final QueryableStoreType<T> queryableStoreType,
                                              final KafkaStreams streams) throws InterruptedException {
  while (true) {
    try {
      return streams.store(storeName, queryableStoreType);
    } catch (InvalidStateStoreException ignored) {
      // store not yet ready for querying
      Thread.sleep(100);
    }
  }
}

整个示例可以在confluent github上找到。