kafka流|州商店(Comp Store)可能已迁移到另一个实例

Question

我是kafka和流的新手。我正在创建一个本地商店，以保留特定主题components的所有更新。我不是在这里做错什么。有其他方法可以从流创建商店吗？

我需要在kafka中创建一个主题 comp-store？

public class MyStream {
final static CountDownLatch latch = new CountDownLatch(1);   
    private static final String APP_ID = "MyTestApp";
    public static void main(String[] args) throws InterruptedException {
        final Properties streamsConfiguration = getStreamsConfiguration();
        final StreamsBuilder builder = new StreamsBuilder();
//        
        final KStream<String, Component> componentStream =  builder.stream("components");
        final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);    
        KeyValueMapper<String, Component, Iterable<KeyValue<String, Component>>> mapper = new KeyValueMapper<String, Component, Iterable<KeyValue<String,Component>>>() {

            @Override
            public Iterable<KeyValue<String, Component>> apply(String list, Component comp) {
                ArrayList<KeyValue<String, Component>> result = new ArrayList<>();
                result.add(KeyValue.pair(comp.getCompId()+":"+comp.getListId(), comp));
                return result;
            }
        };
        KStream<String,Component> componentsStram = componentStream.flatMap(mapper);
        KGroupedStream<String,Component> componentsGroupedStream = componentsStram.groupByKey();
        componentsGroupedStream.reduce(new Reducer<Component>() {
            public Component apply(Component oldVal, Component newVal) {
                return newVal;
            }
        }, Materialized.<String, Component, KeyValueStore<Bytes, byte[]>>as("comp-store"));

        streams.start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(true){
                     if(streams.state().isRunning()){
                            latch.countDown();
                        }
                }
            }
        }).start();
        latch.await();
        Thread.sleep(5000);
        ReadOnlyKeyValueStore<String,Component> localStore = waitUntilStoreIsQueryable("comp-store", QueryableStoreTypes.<String, Component> keyValueStore(), streams);
        System.out.println(localStore.approximateNumEntries());
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
    private static Properties getStreamsConfiguration() {
        Properties settings = new Properties();
        settings.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
        settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        settings.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        settings.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, ProtoSerde.class);
        settings.put(StreamsConfig.STATE_DIR_CONFIG, "C:\temp");
        settings.put("auto.offset.reset","earliest");
        settings.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        return settings;
    }
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) {
            Thread.sleep(100);
        }
    }
}

}

异常

 Exception in thread "main" org.apache.kafka.streams.errors.InvalidStateStoreException: The state store, comp-store, 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:1038)
at com.mr.streams.MyStream.main(MyStream.java:110)

update waitUntilStoreIsQueryable后，我的例外得到解决，但我仍然无法查询状态存储。看来它是在无限循环中进行的。但是，数据中存在于componentsStram中。我在这里做错了吗？

Answer 1

异常被抛出，因为kafka streams实例尚未准备就绪。

根据文档：https：//docs.confluent.io/current/current/streams/faq.html#interactive-queries，可能有两个原因：

• 本地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);
    }
  }
}

更新：

定义整个拓扑后，您必须移动Kafkastream的创建：线：final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);应该追随：componentsGroupedStream.reduce(...)