Потоки Кафки: как обрабатывать динамические условия в фильтре?

Question

Есть ли способ обработать динамическое условие в потоке кафки? Мне нужно отфильтровать мои потоковые данные по списку значений, измененных пользователями, но этот список значений не находится в потоке, он доступен через вызов http.

stream(myTopic)
    .filter(isDataOK())
    ...

private Predicate<> isDataOK() {
     http_call;
     return predicate_value_based_on_http_answer;
}

Можно ли обрабатывать вызов http во время обработки потока kafka или мне нужны условные данные в другом потоке?

Спасибо заранее, Привет

Answer 1

У меня были похожие требования к конфигурации динамического потока. Вызов http для каждой обработки сообщений не очень хорошая идея, так как это отнимает много времени. Лучше кэшировать значения в компоненте обработки и обновлять их регулярно, например, один раз в час. Вам также следует подумать о сценарии, когда http-вызов недоступен.

В другом проекте я получал динамическую конфигурацию асинхронно из другой темы Kafka, и это работало лучше и более естественно для потоковой обработки.