Вариант использования - очистить записи из разделов, которые не получили новые данные в потоках kafka, поскольку мы используем подавление, которое требует времени потока.
Итак, у нас есть Window Store с окном переворачивания в 1 минуту с операцией уменьшения, присоединенной с подавлением. Конструкция подавления зависит от времени потока. Таким образом, если какой-либо раздел не получает новую запись потребителя, подавление не будет выполняться для ожидающей записи потребителя в этом разделе.
Стоит отметить, что срок хранения государственного хранилища установлен на 65 секунд.
Итак, чтобы выполнить явный сброс из хранилища состояний окна, было решено использовать tranform api и использовать его в топологии DSL.
В узле преобразования мы используем context.schedule для планирования пунктуатора, чтобы получить доступ к хранилищу состояний и запустить оконный запрос, то есть fetchall (startTimeInstant, endTimeInstant), чтобы получить старые ключи, которые еще не сброшены.
Из документации стоит отметить, что срок хранения - это минимальное время, в течение которого данные будут оставаться в хранилище окон. Только если все записи в окне достаточно старые, удаляется только оно.
Теперь идея состоит в том, что успешные записи не должны находиться в хранилище состояний, когда мы запускаем fetchall (время начала равно (utc-3minutes). Но до 6 минут старые данные, которые были удалены, все еще находятся в хранилище окон.
ПРОБЛЕМА здесь заключается в том, что я не хочу видеть старые записи в хранилище окон, так как в этом случае необходимо просмотреть / проанализировать полезную нагрузку, чтобы сделать выбор, очищать данные или нет, что требует больших затрат производительности.
Я также проверил политику сжатия / удаления тем в хранилище журналов изменений. Он также имеет 65 секунд.
Я знаю, что классический подход заключается в отправке пакета keep alive на все разделы входной темы, но в нашем случае это невозможно, поскольку входная тема используется несколькими клиентами. Им всем придется измениться.
