Кафка Стрим |Хранилище состояний comp-store могло быть перенесено в другой экземпляр - PullRequest
Новая
       17

Кафка Стрим |Хранилище состояний comp-store могло быть перенесено в другой экземпляр

0 голосов
/ 09 мая 2019

Я новичок в Кафке и стрим.Я создаю локальный магазин для хранения всех обновлений из определенной темы components.Я не понимаю, что я делаю здесь неправильно.Есть ли другой способ создать магазин из Stream?

Нужно ли создавать тему 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)

Обновление После waitUntilStoreIsQueryable мое исключение разрешено, но я все еще не могу запросить хранилище состояний.Кажется, это происходит в бесконечном цикле.Однако данные есть в componentsStram.Я что-то здесь не так делаю?

1 Ответ

2 голосов
/ 09 мая 2019

Исключение выдается, потому что экземпляр Kafka Streams еще не готов.

Согласно документации: https://docs.confluent.io/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);
    }
  }
}

UPDATE

Вы должны переместить создание KafkaStream после определения всей топологии: строка: final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration); должна идти после: componentsGroupedStream.reduce(...)

...