Блог

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

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

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

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

Заранее спасибо, с уважением

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

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