Я тоже наблюдал подобные сбои и на других конфигурациях и на этом docker-образе kafka в целом. На постоянной основе не воспроизводится, поэтому разобраться тоже пока не удалось.
На вопрос, когда такой подход целесообразно применить, гораздо лучше меня ответит список сценариев использования библиотеки Parallel Consumer: https://github.com/confluentinc/parallel-consumer/blob/master/README.adoc (пункт 3.4. Scenarios). В основном, насколько я понимаю, авторы Parallel Consumer предлагают использовать этот подход, когда увеличивать количество партиций не представляется возможными, либо когда увеличение партиций значительно не ускоренит обработку сообщений.
Потеря сообщений после падения приложения
Потери сообщений можно избежать, если фиксировать пакет прочитанных сообщений явно вызовом метода Commit в коде обработчика сообщений только после того, как сообщения обработаны. Если приложение упадет в процессе обработки пачки сообщений, то после перезапуска приложения все незафиксированные сообщения считаются consumer'ом повторно.
Тут еще надо сказать, что segmentio/kafka неявно для клиента библиотеки считывает из kafka сообщения в буферный канал msgs, и именно из msgs извлекается по одному сообщения вызовом метода Fetch. Думаю, что большинство реализаций consumer поступают аналогично - используют буфер считанных сообщений в оперативной памяти приложения.
Менее активное использование одного канала по сравнению с остальными (если я правильно вас понял).
Думаю, что это проблема не подхода с каналами, а больше проблема балансировки нагрузки и распределения сообщения по очередям. Ведь может оказаться, что и одна из партиций kafka будет пуста, потому что алгоритм распределения сообщений не гарантирует равномерного распределения.
Я согласен, что в любом подходе есть плюсы и минусы. Думаю, что нужно находить компромиссные решения с учетом исходных данных и ограничений решаемой задачи.
Хотелось показать, как можно действовать в условиях, когда есть возможность использовать только ограниченное количество партиций. Пример, конечно, сильно упрощенный. Параллельный consumer имеет смысл использовать, когда обработка сообщения заключается в обращении к нескольким внешним системам и применении проверок предметной области, а не только в записи в базу.
Такие эксперименты не продовил. Попробую порассуждать.
Насколько мне известно, producer всегда записывает сообщения в Leader-партицию.
Пусть producer посылает сообщение в партицию 0. Запись в партицию 0 в данный момент недоступна. Пусть запись недоступна по следующей причине. У consumer установлен параметр acks=all, сообщение записалось в Leader-партицию, и не записалось в необходимое количество Follower-партиций (запись подтвердили меньше Follower-партиций, чем установлено в параметре брокера min.insync.replicas). Тогда сообщение, которое producer пытался записать в партицию 0 не будет записана ни в какую другую партицию - producer получит ошибку.
Получается, что при записи сообщение не попадет в неправильную партицию
Теперь посмотрим на чтениие.
По умолчанию consumer читает сообщения из Leader-партиции. Пусть чтение из Leader-партиции невозможно. Такое могло произойти из-за сетевых задержек или перегрузки брокера. Тогда из Follower-партиций назначается новая Leader-партиция, и из нее продолжит читать сообщения consumer. Если новая Leader-партиция будучи Folower-партицией не успела получить все сообщения от прежней Leader-партиции, то consumer не получит часть сообщений. Насколько я знаю, эти сообщения kafka не восстановит.
Я тоже наблюдал подобные сбои и на других конфигурациях и на этом docker-образе kafka в целом. На постоянной основе не воспроизводится, поэтому разобраться тоже пока не удалось.
Балансер находиться в продюсере.
Целесообразность подхода
На вопрос, когда такой подход целесообразно применить, гораздо лучше меня ответит список сценариев использования библиотеки Parallel Consumer: https://github.com/confluentinc/parallel-consumer/blob/master/README.adoc (пункт 3.4. Scenarios).
В основном, насколько я понимаю, авторы Parallel Consumer предлагают использовать этот подход, когда увеличивать количество партиций не представляется возможными, либо когда увеличение партиций значительно не ускоренит обработку сообщений.
Потеря сообщений после падения приложения
Потери сообщений можно избежать, если фиксировать пакет прочитанных сообщений явно вызовом метода Commit в коде обработчика сообщений только после того, как сообщения обработаны. Если приложение упадет в процессе обработки пачки сообщений, то после перезапуска приложения все незафиксированные сообщения считаются consumer'ом повторно.
Тут еще надо сказать, что segmentio/kafka неявно для клиента библиотеки считывает из kafka сообщения в буферный канал msgs, и именно из msgs извлекается по одному сообщения вызовом метода Fetch. Думаю, что большинство реализаций consumer поступают аналогично - используют буфер считанных сообщений в оперативной памяти приложения.
Менее активное использование одного канала по сравнению с остальными (если я правильно вас понял).
Думаю, что это проблема не подхода с каналами, а больше проблема балансировки нагрузки и распределения сообщения по очередям. Ведь может оказаться, что и одна из партиций kafka будет пуста, потому что алгоритм распределения сообщений не гарантирует равномерного распределения.
Я согласен, что в любом подходе есть плюсы и минусы. Думаю, что нужно находить компромиссные решения с учетом исходных данных и ограничений решаемой задачи.
Надо было больше внимания в статье уделить ограничениям, тогда, возможно, решения выглядели бы более оправданными. Учту это в будущем.
Спасибо вам за комментарий.
Мне хотелось рассмотреть решения связанные именно с kafka. Ведь kafka довольно часто применяют для решения подобных задач.
Согласен с вами, есть более оптимальные решения.
Хотелось показать, как можно действовать в условиях, когда есть возможность использовать только ограниченное количество партиций. Пример, конечно, сильно упрощенный. Параллельный consumer имеет смысл использовать, когда обработка сообщения заключается в обращении к нескольким внешним системам и применении проверок предметной области, а не только в записи в базу.
Такие эксперименты не продовил. Попробую порассуждать.
Насколько мне известно, producer всегда записывает сообщения в Leader-партицию.
Пусть producer посылает сообщение в партицию 0. Запись в партицию 0 в данный момент недоступна. Пусть запись недоступна по следующей причине. У consumer установлен параметр acks=all, сообщение записалось в Leader-партицию, и не записалось в необходимое количество Follower-партиций (запись подтвердили меньше Follower-партиций, чем установлено в параметре брокера min.insync.replicas). Тогда сообщение, которое producer пытался записать в партицию 0 не будет записана ни в какую другую партицию - producer получит ошибку.
Получается, что при записи сообщение не попадет в неправильную партицию
Теперь посмотрим на чтениие.
По умолчанию consumer читает сообщения из Leader-партиции. Пусть чтение из Leader-партиции невозможно. Такое могло произойти из-за сетевых задержек или перегрузки брокера. Тогда из Follower-партиций назначается новая Leader-партиция, и из нее продолжит читать сообщения consumer. Если новая Leader-партиция будучи Folower-партицией не успела получить все сообщения от прежней Leader-партиции, то consumer не получит часть сообщений. Насколько я знаю, эти сообщения kafka не восстановит.
Тогда отвечу на ваш вопрос так: не повлияют.
Но, повторюсь, собственноручно я это не проверял.