да, по ходу там какие то сетевые проблемы были. у меня инет через симку и когда пинг становится до хоста трехзначный (500+мс) консамеры тупо сгребать данные перестают хотя выглядят как подключенные.

я пока стратегию построил так: часть консамеров в группе стартует с начала и до того момента которые определены как оффсет на момент подключения консамеров из группы Б, параллельно включаю консамеры из группы А но не с первого оффсета,а с последнего. итого на момент заканчивания работы первой группы я имею условно актуальные данные. то есть далее актуальность поддерживают консамеры из группы А.

всё.

вот каунтеры сделал да на конкурентхашмапе, в которой по ключам каждый консамер сохраняет последний оффсет. но кейса когда разные консамеры дергают один и тот же ключь улсовно "параллельно" - у меня нет.

что именно через жопу? я сталкиваюсь всё чаще с кейсами когда задача чуть сложнее перетаскивания джейсона из угла в угол и всё все горят. типа слооожна.

еще раз медленно.

надо в бд слить с топика данные. слить сначала. и бесшовно бд передать на поддержку другому топику.
первый топик компакт второй не компакт. между топиками есть разумный лаг но и тот и этот получают одно и то же.

как я это по крайней мере пока сделал:

запускаем с бегиннинга консамеры на топик1, при подключении все консамеры дружно запоминают максимальный оффсет на момент подключения. (та самая конкурентная мапа где ключ - партиция, значение - максОффсет на момент снятия данных. т.е. если я даже следом добавлю еще консамеров, или они отвалятся - ничего не произойдет они просто ребалансятся сами а данные берут оттуда же и доступ по ключам между консамерами конкурентный в принципе исключен. каждый консамер мутирует только значение своего ключа или ключей).

параллельно запускаем консамеры на топик2, но уже с КОНЦА (т.е. лови сообщения созданные после подключения). ждем пока консамеры топика1 дойдут до максимального оффсета, и гасим их. бд синхронизирована. топик2 консамеры продолжают работать. бд в состоянии синхрона.

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

зачем два топика? ведь можно было бы и дальше содержать только топик1? отвечаю. в топике1 нет событий по удалению записей. они через томбстон оттуда просто "исчезают" по ключам и эти события прилетают только через топик2. поэтому топик1 годен только для инициальной заливки бд, но не годен для постоянного синхрона. топик2 же наоборот имеет вменяемый ретеншн типа 2 недели и с него ты не вытянешь все данные для инициализации.

почему так сложно? а вот так вот. в идеале хорошо иметь один топик. и одну консамергруппу. но в данном случае это невозможно потому что нету поставщиков данных.

а если базы НЕТ? тут чуть ранее все топили 1 бд 1 сервис и всё остальное на ивентах. получите распишитесь. а то как то непоследовательно.

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

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

максимальный оффсет нужен чтоб знать - залились ли данные в промежутке от сих и до сих или нет. оффсет это что то что всегда растет и никогда не падает (по крайней мере в кафке).

еще раз. речь идет НЕ о репликации, речь идет о ремаппинге данных как минимум. они не кладутся в базу в том виде в каком летят из топика. то есть о подготовке и хранении в ином формате, считай скрс.

почти. только читать начинаю с обоих топиков одновременно.
у одного топика настройка ирлест у второго латест.

итого таймлайн 1,2,3,4,5,6,7,8,9,10... итп. у нас точка входа типа в 7 (это таймлайн, это НЕ оффсеты), где 7 это настоящее время, 1 - это далекое прошлое и 10 это недалекое будущее. так вот, с топика А (который не содержит удаленных записей) мы читаем с 1 по 7. ,
с топика Б мы читаем с 7 - и дальше в будущее. читание начинается одновременно. то есть до того как топик А выключится у нас 1,7
12,78 123,789 и т.п. и в какой то момент времени становится 1,2,3...10. 1,2,3...11. 1,2,3...12 и т.п. топик А же выключится тогда когда гарантированно построится картина 1,2,3,4,5,6,7 (дальше не важно).

никаких комплектов нет. синхронизаций нет буферов нет. говорю же. есть два топика - один компакт который содержит все ФАКТИЧЕСКИЕ записи, второй с ретеншеном, который содержит события удаления записей.

первый топик это типа "исторический" второй - текущий.

короче, я всё сделал. выглядит не очень, но как минимум, все консамеры стейтлесс и не должны друг о друге ничего знать и не знают.

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

ты вошел в топик и говоришь кафке - покажи мне ласт оффсеты по всем партициям. понятно что на момент выдачи тебе этих значений они могут быть уже и устаревшими. но как минимум они очень близки к неким "реальным". хотя что такое реальность?

это выглядит так (по крайней мере внешне) что полл дернул но ничего не пришло.
те конфиги что ты показал я естественно заиспользовал. размеры мессаджей там несоклкьо сотен килбоайт. батч на полл как раз 500.
поллит раз в полсекунды.

я поигрался с настройками, ты ставишь батч 500 тебе с 10-ти партиций прилетает только в 1-2. ставишь батч 50 - тебе прилетает со всех 10ти партиций. :) но.. оптяь же оно может и по 500 с 10-ти партиций слать, но гораздо реже.