Приостановить потребителя Camel, если значительное количество сообщений последовательно не работают

Question

Я хотел бы приостановить потребителя Camel Kafka (т.е. маршрут) на основе результата обработки последнего сообщения . Например, когда 10 последних завершенных сообщений не удалось выполнить с той же ошибкой, я хотел бы приостановить использование сообщений вместо отправки тысяч сообщений с ошибкой topi c.

Я обнаружил, что RoutePolicies like ThrottlingInflightRoutePolicy могут приостановить маршрут на основе текущих обменов во время полета путем доступа к InflightRepository.

Итак, мой вопрос закипает вплоть до: есть ли также возможность для просмотра последних n завершенных сообщений маршрута? Думаю, это позволило бы мне реализовать RoutePolicy для моего случая.

Или, если нет, могу ли я построить это средство самостоятельно, просто построив коллекцию сводок обработки последних нескольких обменов в поле экземпляра RoutePolicy?

Спасибо за любые подсказки.

Answer 1

Да, для этого нет встроенной политики маршрутизации. У нас есть, как вы говорите, дросселирование, а также для cron, например, планирование.

Таким образом, вы можете создать свое собственное и сохранить окно с последними N сообщениями и отслеживать, сколько из них не удалось, а затем приостановить, когда вы достигнете этого порога.

Имейте в виду, что насчет возобновления снова? Если это произойдет автоматически, например, через 5 минут, попробуйте еще раз. В противном случае это ручной шаг, чтобы заставить потребителя снова возобновить работу.