А такой вопрос - реальна ли ситуация, что консюмер начнет

читать с earliest на грани с ttl сообщений, и перед вторым poll сработает ретеншен, который заденет так же данные которые должны были считаны вторым poll, и консюмер потеряет какие то данные ?

Конечно реальна.

и как с подобным живут, понимаю что кейс 1 на миллион, но как с этим жить)?

А в чём проблема-то? Начни он читать на минуту позже - он бы их и так потерял.

предположим, что мы сложили историю топика из с3 в таблицу, в ней ключ X имеет значение 10 (X = 10) потом запустили процесс который из кафки обновляет эту таблицу, он начинает читать с самого начала - тк первый раз к ней подключается, он прочитал из первого poll - что ключ X=1, применил, (мы надеемся на eventually consistency, что в конечном итоге мы согласуемся с источником) во втором poll содержится что X=10, но мы его не получили, В итоге X = 1 , и останется им - пока X кто нибудь не обновит еще раз Можно придумывать всякие if else и применять обновления только если offset больше записанного, но не факт что в с3 будет оффсет или что ваша пишущая коробка в виде синк коннектора поддерживает такую логику, ну и в целом это не оптимально

И отсутствующая запись при этом допустима? Просто если мы говорим о согласованности с источником, то потери вообще недопустимы, соответственно работа на границе ретеншна - не вариант, либо ретеншен побольше нужен, либо ещё что-то. Я могу представить кейс, когда потеря записи допустима, а некорректная запись - нет, но это уже реально особый случай и проще его решать в рамках существующих ограничений (т.е. пойти в сторону более строгих гарантий, как в предыдущем случае).

Выглядит как будто вам тогда нужен compacted топик или долгий ретеншн

что мешает засетапить топик с бесконечным ретеншном, стартануть обработку, после прохода уже настроить нужный ретеншн уже с учетом рисков и скорости обработки?