Kafka Streams 持久存储错误:状态存储，可能已迁移到另一个实例 [英] Kafka Streams persistent store error: the state store, may have migrated to another instance

问题描述

我在 Spring Boot 中使用 Kafka Streams.在我的用例中，当我从其他微服务接收客户事件时，我需要存储在 customer 物化视图中，当我收到 order 事件时，我需要加入客户并订购然后存储在客户订单物化视图中.为了实现这一点，我创建了持久键值存储 customer-store 和 当有新事件发生时更新它.

I am using Kafka Streams with Spring Boot. In my use case when I receive customer event from other microservice I need to store in customer materialized view and when I receive order event, I need to join customer and order then store in customer-order materialized view. To achieve this I created persistent key-value store customer-store and updating this when a new event comes.

    StoreBuilder customerStateStore = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("customer"),Serdes.String(), customerSerde).withLoggingEnabled(new HashMap<>());
streamsBuilder.addStateStore(customerStateStore);
KTable<String,Customer> customerKTable=streamsBuilder.table("customer",Consumed.with(Serdes.String(),customerSerde));
    customerKTable.foreach(((key, value) -> System.out.println("Customer from Topic: "+value)));

配置的拓扑、流和启动的流对象.当我尝试使用 ReadOnlyKeyValueStore 访问存储时，我收到以下异常，即使我不久前存储了一些对象

Configured Topology, Streams and started streams object. When I try to access store using ReadOnlyKeyValueStore, I got the following exception, even though I stored some objects few moments ago

streams.start();
ReadOnlyKeyValueStore<String, Customer> customerStore = streams.store("customer", QueryableStoreTypes.keyValueStore());
System.out.println("customerStore.approximateNumEntries()-> " + customerStore.approximateNumEntries());

代码上传到 Github 以供参考.感谢您的帮助.

Code uploaded to Github for reference. Appreciate your help.

异常:

org.apache.kafka.streams.errors.InvalidStateStoreException: the state store, customer, may have migrated to another instance.
    at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:60)
    at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1043)
    at com.kafkastream.service.EventsListener.main(EventsListener.java:94)

推荐答案

状态存储通常需要一些时间来准备.最简单的方法如下.(代码来自官方文档)

The state store needs some time to be prepared usually. The simplest approach is like below. (code from the official document)

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);
    }
  }
}

您可以在文档中找到更多信息.https://docs.confluent.io/current/streams/faq.html#interactive-queries

You can find additional info in the document. https://docs.confluent.io/current/streams/faq.html#interactive-queries

这篇关于Kafka Streams 持久存储错误:状态存储，可能已迁移到另一个实例的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！