InvalidStateStoreException при соединении KStream с использованием GlobalKtables - PullRequest
Новая
       33

InvalidStateStoreException при соединении KStream с использованием GlobalKtables

1 голос
/ 06 мая 2019

У меня есть приложение Kafka Streams, где я присоединяюсь к KStream, который читает из «topic1», с GlobalKTable, который читает из «topic2», а затем с другим GlobalKTable, который читает из «topic3».

Когда япопробуйте отправить сообщения во все 3 темы одновременно, тогда я получу следующее исключение -

org.apache.kafka.streams.errors.InvalidStateStoreException

Если я нажмусообщения одно за другим в этих темах, т. е. отправлять сообщения в теме 2, затем в теме 3, а затем в теме 1, тогда я не получаю это исключение.

Я также добавил StateListener перед тем, как запустить KafkaStreams 

KafkaStreams.StateListener stateListener = new KafkaStreams.StateListener() {
            @Override
            public void onChange (KafkaStreams.State newState, KafkaStreams.State oldState) {
                if(newState == KafkaStreams.State.REBALANCING) {
                    try {
                        Thread.sleep(1000);
                    }
                    catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
streams.setStateListener(stateListener);
streams.start();

Кроме того, я жду, пока хранилище не будет запрошено после запуска потока, вызывая следующий метод 

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 (final InvalidStateStoreException ignored) {
                // store not yet ready for querying
                Thread.sleep(100);
            }
        }
    }

Ниже приведен код соединения Kafka Streams и GlobalKTable: 

KStream<String, GenericRecord> topic1KStream =
           builder.stream(
               "topic1",
               Consumed.with(Serdes.String(), genericRecordSerde)
           );
GlobalKTable<String, GenericRecord> topic2KTable =
           builder.globalTable(
               "topic2",
               Consumed.with(Serdes.String(), genericRecordSerde),
               Materialized.<String, GenericRecord, KeyValueStore<Bytes, byte[]>>as("topic2-global-store")
                   .withKeySerde(Serdes.String())
                   .withValueSerde(genericRecordSerde)
           );
  GlobalKTable<String, GenericRecord> topic3KTable =
           builder.globalTable(
               "topic3",
               Consumed.with(Serdes.String(), genericRecordSerde),
               Materialized.<String, GenericRecord, KeyValueStore<Bytes, byte[]>>as("topic3-global-store")
                   .withKeySerde(Serdes.String())
                   .withValueSerde(genericRecordSerde)
           );

KStream<String, MergedObj> stream_topic1_topic2 = topic1KStream.join(
           topic2KTable,
           (topic2Id, topic1Obj) -> topic1.get("id").toString(),
           (topic1Obj, topic2Obj) -> new MergedObj(topic1Obj, topic2Obj)
       );
       final KStream<String, GenericRecord> enrichedStream =
        stream_topic1_topic2.join(
           topic3KTable,
           (topic2Id, mergedObj) -> mergedObj.topic3Id(),
           (mergedObj, topic3Obj) -> new Enriched(
               mergedObj.topic1Obj,
               mergedObj.topic2Obj,
               topic3Obj
           ).enrich()
       );
enrichedStream.to("enrichedStreamTopic", Produced.with(Serdes.String(),getGenericRecordSerde()));

Приведенный выше кодочень похоже на this .

Когда я пытаюсь отправить сообщения во все 3 темы одновременно, я получаю следующее исключение: 

org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_1, processor=KSTREAM-SOURCE-0000000000, topic=topic1,
partition=1, offset=61465,
stacktrace=org.apache.kafka.streams.errors.InvalidStateStoreException:
Store topic2-global-store is currently closed.
    at
org.apache.kafka.streams.state.internals.WrappedStateStore.validateStoreOpen(WrappedStateStore.java:66)
    at
org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:150)
    at
org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:37)
    at
org.apache.kafka.streams.state.internals.MeteredKeyValueStore.get(MeteredKeyValueStore.java:135)
    at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl$KeyValueStoreReadOnlyDecorator.get(ProcessorContextImpl.java:245)
    at
org.apache.kafka.streams.kstream.internals.KTableSourceValueGetterSupplier$KTableSourceValueGetter.get(KTableSourceValueGetterSupplier.java:49)
    at
org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:71)
    at
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
    at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
    at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
    at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
    at
org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
    at
org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
    at
org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
    at
org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
    at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
    at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
    at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)

1 Ответ

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

Я исправил проблему в своем коде, у меня было auto.register.schemas=false, потому что я вручную зарегистрировал схемы для всех своих тем.

После того, как я установил auto.register.schemas=true и перезапустил приложение потоков, оно заработало нормально.Я думаю, что он нуждается в этом флаге для своих внутренних тем.

...