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

ожидаемое поведение - стартуем 10 консамеров - все 10 консамеров гребут по 1000 сообщений.
что в реале - из 10-ти консамеров рандомно 2-3 консамера выгребают от 200 и до 1000 сообщений за один цикл

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

Так в логах ниче нет. Полл прошел консумер выгреб ноль.
Групайди одна. Это прям 143%.
Выглядит так будто кафка типа "не готова" Отдавать сообщения, А в следующий полл может быть готова. А может тоже нет. Я то думал там можно до бесконечности накидать партиций и Консамеров. И оно типа будет работать. Но вот оказывается нет :)

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

в том и дело что я читаю топик сначала. а оно отдает сообщения порциями с невменяемыми паузами.

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

в общем, судя по тому как валятся мессаджи я думаю всё же какие то моменты с конфигурацией самой кафки.

фиг с ним. спрошу как всегда девопсов что там наворотили.

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

ок.

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

как я это вижу. я вначале собираю в какую нибудь конкурентную мапу ключами все партиции (1-2-3-4-5-6 и т.п.) и при каждом обработанном рекорде вытаскиваю текущий оффсет и максимальный оффсет (на момент получения рекорда) и кладу в мапу по ключу партиции значение разницы между ними.

в итоге у меня получается некий реестр с партициями и остатком мессаджей. и вот я периодически смотрю когда там всё будет по нулям (на самом деле по единицам) - я гашу сервис.

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

есть еще какие то варианты дочитать топик до условного конца и выключиться?

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

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

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

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

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

я понимаю что "конец" очереди это вещь весьма условная. потому что он может быть конец на определенный момент времени который у всех к тому же разный. я думаю что конец а за долю секунды до того или после того залетело уже новое сообщение а я не знаю и думаю что конец а оно прилетает. понятно что это всё ОЧЕНЬ условно.
поэтому цель стоит прочитать всё что нужно из топика до определенного оффсета,потому что после него уже читают другие. или незадолго до него.
оффсет это понятие абсолютное. вот когда ты его попросил у кафки оно посмотрело в партицию и отдало тебе это значение. всё. так же у партиции есть как минимум, начало.

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

да, по ходу там какие то сетевые проблемы были. у меня инет через симку и когда пинг становится до хоста трехзначный (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-ти партиций слать, но гораздо реже.

Вимба, у меня одно сообщение хорошо если килобайт есть - скорее меньше.

я конечно ща может фигню скажу, но по ходу если пинги до хоста где кафка начинают заваливаться за 200-300мс -- то оно начинает слать сильно меньше и реже или вообще в нулину уходит. НО при этом всякие коннекшн эрроры не прилетают. если под 500 и потери пакетов - тогда да. всякие ошибки фетча и коннекшина. но это уже совсем край и рассматривать не стоит.