Есть ли готовые решения для очереди с десятком тысяч заранее не заданных партиций?

Question

[ugin_root]

Проблема: нужно синхронизировать мои данные со сторонними сервисами.

Дано:

1. Мои данные ClientId, CustomerId, CustomerEmail, CustomerPoints
   
2. Четыре (в перспективе до 20) сервиса с которыми нужно синхронизировать мои данные
   
3. Примерно 15 000 клиентов (ClientId), у клиентов от 1000 до 500 000 customer-ов
   
4. У каждого сервиса есть rate-limit. Доступ к каждому сервису для каждого клиента осуществляется по приватному ключу, т.е. rate-limit индивидуальный для связки ClientId-ServiceName
   

Текущее решение:

1. Есть kafka со 100 партициями, партиция это ClientId % 100 = (0 - 99)
   
2. При обновлении данных customer-а в kafka topic пишется сообщение о том чей он клиент (ClientId) и какой у него id
   
3. Проверяется с какими сервисами у клиента настроена синхронизация
   
4. Запускается параллельно синхронизация со всеми активными сервисами
   
5. При неудаче синхронизация повторяется до тех пор пока не выполнится (rate-limit и отказы сервисов учитываются именно сдесь)
   

Какие сейчас есть проблемы:

1. При подключении нового сервиса клиентом, мне необходимо синхронизировать всех его customer-ов с этим сервисом, а их у него может быть до 500 000. Партиция забьётся и те клиенты которым не повезло оказаться с ним в одной партиции будут очень долго синхронизироваться учитывая все rate-limit-ы.
   
2. При достижении rate-limit-а у определённого клиента на каком либо из сервисов, страдает синхронизация для всех остальных сервисов и для всех клиентов сообщения от которых тоже находятся в этой партиции.
   
3. Если сторонний сервис выходит из строя на 20 минут, то каждая партиция в которой появится сообщение для клиента данные которого необходимо синхронизировать с этим сервисом "зависнет" на 20 минут.
   

Какое решение для очереди сообщений мне нужно реализовать (найти готовое):

1. Неограниченное количество партиций
   
2. Партиция должна быть строкой или составным значением (ClientId, ServiceName)
   
3. Каждая партиция должна обрабатываться конкретным consumer-ом
   
4. Когда в партиции заканчиваются данные она пропадает и consumer переключается на обработку другой
   
5. Возможность повторной обработки сообщений в случае ошибок
   
6. Проверка есть ли в очереди сообщение по его ключу и значению (это не обязательно, но желательно что-бы избежать дубликатов)
   
7. Проверка размера необработанных сообщений в определённой партиции (это не обязательно, но желательно)
   

Сразу отброшу kafka т.к. нет возможности заранее узнать все партиции и держать огромное их количество.
Рассматривал Apache Pulsar, но там количество партиций задаётся глобально в сетингах что тоже не подходит, или я невнимательно изучал этот вопрос, если так то поправьте.
RabbitMQ совсем как мне кажется про другое.
В голову лезут мысли про свой собственный велосипед поверх:

1. redis streams
   
2. redis hset+list
   
3. mongodb (но тут точно возникнут проблемы с выполнением на нескольких серверах)
   
4. sqs + в памяти разбивать по очередям для каждого ClientId+ServiceName (плохо параллелится и возможны проблемы по памяти)
   

Может у кого-то были схожие задачи и есть готовые реализации подобных очередей?

Comments

[mayton2019]

Что значит

Когда в партиции заканчиваются данные она пропадает и consumer переключается на обработку другой

?

Как узнать что в стриминговой системе вдруг закончились данные?

[Shavadrius]

Не пробовали более разветвленную схему топиков? Например, сделать отдельные для операций синхронизации, добавления и удаления сервисов/подписок.

[ugin_root]

mayton2019, Не совсем корректно выразился. Партиция не пропадает, в ней просто в данный момент нет данных и нужно переключиться на другую.

[ugin_root]

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

[mayton2019]

ugin_root, я не совсем понимаю что значит "переключится" на другую.
Может быть ты имеешь в виду диспетчеризацию внутри процесс-консюмера?

Например я - процесс потребитель событий. И с одной стороны я могу подключаться к 10 топикам Kafka.
Сливать все 10 в один внутренний для обработки. Те события которые удалось обработать - коммитить
в кафку (да там есть режим фиксации транзакций). А те которые еще не готовы к обработке
я буду вращать по кругу во внутреннем буфере. Какое-то время. 5-10 минут. Выбери сам.
Потом сказать Кафке rollback. Дескыть пока не судьба. Положу обратно на полочку.

[ugin_root]

mayton2019, У меня уже итак всё в одном топике и все сообщения уже готовы к обработке и это последнее что с ними нужно сделать. По сути мне просто нужно выполнить 1 запрос к стороннему сервису и если он выполнится успешно то сообщение считается обработанным. Не понимаю как мне тут могут помочь транзакции.

Kafka бы мне идеально подошла если бы я мог создать порядка 100 000 партиций в топике, а в перспективе мне могут понадобиться ещё несколько таких топиков, но вроде у kafka есть глобальный лимит на количество партиций.

Если бы у меня было 30 клиентов и 3 сервиса я бы создал 90 партиций и указал в параметре partitionsConsumedConcurrently например 10 , то сообщения из каждой партиции обрабатывались бы параллельно и с примерно одинаковой скоростью не "мешая" другим партициям. Если бы я в одну партицию я записал миллион сообщений, это бы не привело к задержкам обработки сообщений из других партиций.

Но к сожалению я не могу создать 100 000 партиций в kafka, или могу?

Я нашёл ещё два варианта которые могли бы решить проблему:
1. Apache Pulsar, но пока что тоже не уверен что он мне подойдёт.
2. AWS SQS FIFO с указанием MessageGroupId - это точно решает мою проблему, но это будет стоить дороже

[mayton2019]

Смотри. Партиции - это инструмент перформанса а не бизнес атрибут. Поэтому если ты сразу сходу
создашь 100 тыщ партиций то тебе надо некоторое количество дисковых хранилищ и серверов
сразу. Беря во внимание что обычно бизнес нагрузка эволюционирует плавно то скорее всего
тебе эти мощности сразу будут не нужны. Но будучи созданными они будут зря тратить деньги
за uptime.

Ты можешь для начала создать 16 партиций а свои топики отобразить на на партиции по ключу (Key).
Кафка поддерживает ключ в каждом месседже поэтому создай себе правильную хеш функцию
которая отображает ключи на партиции. Это в кафке заложено. Вот. Если мощности не хватит - тогда
сделай 32 партиции. Потом 64 и так далее. Вот это и будет правильный подход.

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

[d'Ivan]

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

[ugin_root]

Я нашёл решение которое подходит под мою задачу, это SQS FIFO.

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

При чём если сообщения полученые первым запросом еще не обработались, то при повторном получении сообщений гарантируется что сообщения придут из другой группы.

Но есть одно неприятное ограничение. При получении сообщений проверяются только первые 20 000 и если среди них нет ни одного из не занятой обработкой группы, то ответ придет пустой.

Тут про ограничение https://docs.aws.amazon.com/AWSSimpleQueueService/...

For FIFO queues, there can be a maximum of 20,000 in flight messages (received from a queue by a consumer, but not yet deleted from the queue). If you reach this quota, Amazon SQS returns no error messages. A FIFO queue looks through the first 20k messages to determine available message groups. 

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


Из этих 2.7 миллионов примерно полтора миллиона от одного клиента из-за этого обрабатавается одна группа, т.к сообщения других клиентов видимо лежат дальше чем первые 20 000

[ugin_root]

Иерокопус Таманский, возможно, но нарисовать мне кажется еще сложнее чем объяснить. Завтра попробую нарисовать. Но возможно проще загуглить "sqs fifo groups" есть видео с примерным объяснением как это работает, а это как раз то что мне нужно.