Kafka Streams / Уменьшить и сохранить журнал

Question

У меня есть вопрос о сохранении журнала, и потоки Kafka Streams сокращают операции ...

У меня есть следующая конфигурация потока ...

builder.
 .stream("topic1", Comsumed.with(Serdes.string(), Serdes.string()))
 .groupByKey(Serialized.with(Serdes.string(), Serdes.string()))
.reduce((val1, val2) -> val2, Materialzed.as("store1"))

Я знаю, что захоронения событий не достигаютуменьшить функцию, поэтому я должен использовать специальный объект, сигнализирующий об удалении, но как он работает, если при сохранении журнала удаляется сообщение, как это можно передать в MaterializedView?

Answer 1

Если вы читаете тему в виде потока, а сохранение журнала обрезает тему, вы не сможете получить информацию об этом.

Личное примечание: я думаю, что вы делаете, это анти-шаблон. И проблемы, которые вы решаете, являются индикатором для этого.

Единственный способ «удалить» события из темы - это, если вы используете сжатую тему и читаете ее как KTable. Удаление происходит только через надгробия, которые KTable получит и обработает соответственно.