Spring Cloud Stream: правильная обработка StreamListener, когда обработка сообщения занимает много времени

Question

Когда StreamListener требуется много времени (дольше, чем max.poll.interval.ms) для обработки сообщения, таким образом, конкретный потребитель занят, и другие новые сообщения будут назначены другим разделам.После того, как время превысит max.poll.interval.ms, произойдет перебалансировка, и такая же ситуация произойдет с другим потребителем.Таким образом, это сообщение будет циркулировать вокруг всех разделов и продолжать загружать ресурсы.

Однако такая ситуация встречается не очень часто, только несколько сообщений занимают много времени, и она неуправляема.

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

Answer 1

Увеличение max.poll.interval.ms не окажет влияния на производительность (за исключением того, что потребуется больше времени для обнаружения потребителя, который действительно мёртв).

Каждый раз, когда вы обрабатываете это, выполняется перебалансировка«плохая» запись наносит гораздо больший урон производительности.

Однако вы можете делать то, что хотите, с пользовательским SeekToCurrentErrorHandler вместе с восстановителем, таким как DeadLetterPublishingRecoverer.Вам также понадобится слушатель перебалансировки для подсчета перебалансировок и некоторый механизм для совместного использования состояния из обработчика ошибок между экземплярами (стандартный хранит только состояние в памяти).

Я думаю, довольно сложно.