Кафка Общий государственный магазин между Kafka DSL KStream Transformers - PullRequest
Новая
       39

Кафка Общий государственный магазин между Kafka DSL KStream Transformers

1 голос
/ 06 ноября 2019

У меня есть топология, в которой я использую Transformer для агрегирования своих объектов, позже в своей топологии я пытаюсь читать из State Store, который использовался в первом Transformer. Не представляется возможным получить доступ к данным. Это потому, что хранилища состояний находятся в разных разделах?

Моя топология выглядит примерно так: 

streamsBuilder.stream("input")
    .transform(new TransformerSupplier1(), "my-store")
    .leftJoin(someKTable, myValueJoiner())
    .flatTransform(new TransformerSupplier2(), "my-store")

В моем TransformerSupplier1 Transformer мое хранилище состояний имеет тип <String, Map<String, Object>>

В моих TransformerSupplier2 Transformer я пытаюсь получить хранилище состояний по ключу, который я использовал для хранения в первом преобразователе, но я получаю ноль, и когда я делаю .all().peekNextKey()ничего не найдено.

Дайте мне знать, если мне нужно добавить больше информации для моих Трансформаторов, и я постараюсь запутать реальную логику. Спасибо

...