Кажется, достаточно простой вопрос для знающих Есть кафка топик, в которую

внешняя команду пишет данные. Из этого топика читает batch spark джоба данные, трансформирует и кладет его в хранилище.
Иногда бывает так, что ребята на внешней стороне поошибке/невнимательности/плавающему багу пишут белеберду в message. Из-за этого, собственно, батч джоба благополучно падает. Но мы точно знаем, что offsets с x + n невалидные, т.е. мы их можем просто скипнуть без угрызения совести.

Текущий способ решения этой проблемы:
- Поскольку в kakfa-chechpoints offset проставляется значением из последней успешной батч джобой, то нужно взять код батчджобы и законментировать всю бизнес логику, тогда - все мессажди будут просто прочитаны из топика, без каких либо обработок. Джоба успешно завершится, и чекпоинт обновится - прописав последний оффсет из топика. Таким образом невалидные сообщения в топике будут скипнуты.
Проблемы текущего способа решения очевидны.

Как можно решить такую ситуацию более разумно, и по феншую?

P.S. Разумеется, кафка кластер менеджится отдельной командой, и быстро/пожеланию параметры кафки не поменять. Если надо менять - то нужно проходить кучи согласований и т.д.

Кастомный десериализатор сообщений? Успех - обработали, нет - пропустили. Как я понял, ситуация с невалидными сообщениями может возникать снова и снова, поэтому, на мой взгляд, разумно это в процессинге и предусмотреть

Кстати, тоже к этой мысли подходил. Все невалидные/нераспаршенные месседжи класть в отдельную карантинную таблицу, потом мониторить что в этой таблице. Спасибо за идею!)