Как я могу использовать kafka KeyValueStore в качестве KTable? - PullRequest
Новая
       31

Как я могу использовать kafka KeyValueStore в качестве KTable?

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

Можно ли использовать KeyValueStore в качестве KTable?KeyValueStore будет заполнен операцией преобразования из другой темы.

Это мой KeyValueStore: 

StoreBuilder<KeyValueStore<String, MyType>> definitionStoreBuilder = Stores.keyValueStoreBuilder(
        Stores.inMemoryKeyValueStore(definitionStore),
        Serdes.String(),
        Serdes.serdeFrom(MyType.class))
        .withCachingEnabled();
builder.addStateStore(definitionStoreBuilder);

KeyValueStore<String, MyType> definitionStore = definitionStoreBuilder.build();

И я хотел бы присоединиться к этому с таким потоком: 

KStream<String, String> alert =
            builder.stream("topicName",
            Consumed.with(Serdes.String(),Serdes.String()))
            .join({magicStep(definitionStore)}, (left, right) -> ....
...