Счётчик слов из Kafka: exactly-once и неубывающие значения

На входе — Kafka-топик, в который непрерывно пишутся слова. Нужно считать, сколько раз встретилось каждое слово, и публиковать текущее значение счётчика в другой топик; каждое вхождение учитывается строго один раз (без потерь и без повторного счёта). Пример: вход foo, bar, foo, bar, … → выход (foo,1), (bar,1), (foo,2), (bar,2), … Значения счётчика для одного слова в выходном топике не должны убывать: повтор допустим (…(foo,1)…(foo,2)…(foo,2)…(foo,4)… — корректно), откат (…(foo,2)…(foo,1)…) — нет. Сервис может упасть и перезапуститься в любой момент, после восстановления свойство должно сохраняться.