Kafka Streams: How to use persistentKeyValueStore

问题:

My code is currently using an InMemoryKeyValueStore, which avoids any persistence to disk or to kafka. I want to use rocksdb (Stores.persistentKeyValueStore) so that the app will reload state from disk. I'm trying to implement this, and I'm very new to Kafka and the streams API. Would appreciate help on how I might make changes, while I still try to understand stuff as I go.

I tried to create the state store here:

StoreBuilder<KeyValueStore<String, LinkedList<StoreItem>>> store =
                Stores.<String, LinkedList<StoreItem>>keyValueStoreBuilder(Stores.persistentKeyValueStore(storeKey), Serdes.String(), valueSerde);

How do I register it with the streams builder?

Existing code which uses the inMemoryKeyValueStore:

   static StoreBuilder<KeyValueStore<String, LinkedList<StoreItem>>> makeStoreBuilder(
            final String storeKey,
            final Serde<LinkedList<StoreItem>> valueSerde,
            final boolean loggingDisabled) {

        final StoreBuilder<KeyValueStore<String, LinkedList<StoreItem>>> storeBuilder =
                Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(storeKey), Serdes.String(), valueSerde);
        return storeBuilder;
    }

I need to ensure that the streams app will not end up missing existing messages in the log topic each time it restarts.

回答1:

How do I register it with the streams builder?

By calling StreamsBuilder#addStateStore().

https://kafka.apache.org/22/javadoc/org/apache/kafka/streams/StreamsBuilder.html#addStateStore-org.apache.kafka.streams.state.StoreBuilder-

See StateStoresInTheDSLIntegrationTest at https://github.com/confluentinc/kafka-streams-examples fro an end-to-end demo application.

回答2:

You use a persistent store the exact some way as an in-memory store. The store takes care of the rest and you don't need to worry about loading data etc. You just use it.