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

В программе две теоретические темы Введение и Базовые основы технологии и практическая тема Установка Kafka. В ней поработаем с технологией руками:

1. Развернём Kafka в самом простом её варианте с одним брокером и одной нодой ZooKeeper.
2. Запишем и прочитаем сообщения, посмотрим в конфиги и увидим, как данные хранятся на диске.

Спикеры курса

• Анатолий Солдатов, Lead Engineer в Авито;
• Александр Миронов, Infrastructure Engineer в Stripe, ex-Booking.

Доступ к курсу придёт после регистрации: https://slurm.io/kafka

Релиз всех тем видеокурса будет 7 апреля. До релиза действует цена предзаказа 40 000 руб. вместо 50 000 руб.

В Kafka топик может содержать множество партиций, между которыми распределяются записи. Партиции это единицы параллелизма. В целом, чем больше партиций, тем выше пропускная способность. Однако есть некоторые факторы, которые стоит учитывать, когда в кластере Kafka много партиций. Рад сообщить, что в Apache Kafka, начиная с версии 1.1.0, было существенно увеличено количество партиций, которые может поддерживать один кластер Kafka с точки зрения деплоймента и доступности.

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

Брокер Kafka по умолчанию выполняет контролируемое отключение, чтобы минимизировать сбои в обслуживании клиентов. Контролируемое отключение проходит следующие этапы. (1) Сигнал SIG_TERM отправляется брокеру для завершения работы. (2) Брокер отправляет запрос контроллеру, уведомляя, что он готов к отключению. (3) Контроллер затем меняет лидеров партиций на этом брокере на других брокеров и сохраняет эту информацию в ZooKeeper. (4) Контроллер отправляет информацию о новых лидерах другим брокерам в кластере. (5) Контроллер отправляет выключающемуся брокеру положительный ответ на запрос, и брокер, наконец, завершает свой процесс. Таким образом это никак не влияет на клиентов, потому что их трафик уже перенаправлен на других брокеров. Этот процесс изображен на Рисунке 1. Заметьте, что шаги (4) и (5) могут происходить параллельно.

Рис. 1. (1) Инициация отключения на брокере 1; (2) брокер 1 отправляет запрос о контролируемом отключении контроллеру на брокере 0; (3) контроллер записывает новых лидеров в ZooKeeper; (4) контроллер отправляет информацию о новых лидерах брокеру 2; (5) контроллер отправляет положительный ответ брокеру 1.

До версии Kafka 1.1.0 во время контролируемого отключения контроллер перемещает лидеров по одной партиции за раз. Для каждой партиции контроллер выбирает нового лидера, параллельно записывает его в ZooKeeper и отправляет нового лидера другим брокерам через удаленный запрос. В этом процессе есть несколько недостатков. Во-первых, у синхронной записи в ZooKeeper более высокое время задержки, что замедляет процесс контролируемого отключения. Во-вторых, отправка нового лидера по партиции за раз добавляет много маленьких удаленных запросов к каждому брокеру, что может вызвать задержку обработки новых лидеров.

Начиная с Kafka 1.1.0 и далее контролируемое отключение происходит значительно быстрее. Во-первых, при записи в ZooKeeper используется асинхронный API. Вместо того чтобы записывать лидера для одной партиции, ждать, пока этот процесс завершится, а потом записывать следующего, контроллер асинхронно размещает лидеров для многих партиций в ZooKeeper, а потом ждет завершения процесса. Это позволяет обрабатывать запросы на пайплайне между брокером Kafka и сервером ZooKeeper и снижает общее время задержки. Во-вторых, информация о новых лидерах отправляется батчами. Вместо одного удаленного запроса для каждой партиции контроллер отправляет один удаленный запрос с лидером от всех затронутых партиций.

Время обработки отказа контроллером также было значительно улучшено. Если случается отказ контроллера, кластер Kafka автоматически выбирает контроллером другого брокера. Прежде чем иметь возможность выбрать лидеров партиций свеженазначенный контроллер должен загрузить состояние всех партиций в кластере из ZooKeeper. Если произошел серьезный сбой контроллера, окно недоступности партиции может длиться столько, сколько продолжается срок действия сессии ZooKeeper, плюс время перезагрузки состояния контроллера. Поэтому сокращение времени перезагрузки состояния улучшает доступность в таком редком случаем. До версии 1.1.0 в Kafka для перезагрузки используется синхронный API ZooKeeper. Начиная с версии 1.1.0 это также заменено на асинхронный API для сокращения времени задержки.

Мы провели тесты для оценки улучшений по времени контролируемого отключения и перезагрузки контроллера. Для обоих тестов мы загрузили набор из пяти узлов ZooKeeper на выделенных серверах.

Для первого теста мы подготовили кластер Kafka с пятью брокерами на отдельных серверах. В этом кластере мы создали 25 000 топиков, в каждом топике по одной партиции и две реплики, в общем получив 50 000 партиций. Таким образом на каждом брокере было 10 000 партиций. Затем мы замерили время, которое понадобилось для контролируемого отключения. Результаты представлены в таблице ниже.

Большую часть улучшений дает исправление затрат на журналирование, при котором проводятся ненужные записи для каждой партиции в кластере при смене лидера одной партиции. Просто исправив затраты на журналирование, мы снизили время контролируемого отключения с 6,5 минут до 30 секунд. Переход на асинхронный API ZooKeeper снизил это время до 3 секунд. Эти улучшения существенно снизили время, необходимое для перезагрузки кластера Kafka.

Для второго теста мы подготовили другой кластер Kafka, состоящий из пяти брокеров, создали 2 000 топиков, в каждом по 50 партиций и одной реплике. В сумме во всем кластере получилось 100 000 партиций. Затем мы замерили время перезагрузки состояния контроллера и увидели 100% улучшение (время перезагрузки снизилось с 28 секунд в Kafka 1.0.0 до 14 секунда в Kafka 1.1.0).

Учитывая эти изменения, на поддержку какого количества партиций вы можете рассчитывать в Kafka? Точное число зависит от таких факторов как допустимое окно недоступности, время задержки ZooKeeper, тип хранения на брокере и т.д. В качестве общего правила мы рекомендуем иметь на каждом брокере до 4 000 партиций и до 200 000 партиций в кластере. Основная причина для лимита на кластере заключается в том, что нужно подстраховаться на тот редкий случай серьезного сбоя контроллера, о котором мы писали выше. Обратите внимание, что другие соображения, связанные с партициями, также применимы, и вам может потребоваться дополнительная настройка конфигурации с большим количеством партиций.

Более подробную информацию вы найдете в KAFKA-5642 и KAFKA-5027.

От редакции: Команда Слёрма готовит подробный видеокурс по Apache Kafka. Спикеры Анатолий Солдатов из Авито и Александр Миронов из Stripe. Вводные уроки уже доступны в личном кабинете. Релиз полной версии курса апрель 2021. Подробности.

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

Привет Хабр, на связи Иван Большаков архитектор интеграционных решений, эксперт департамента разработки ПО КРОК. Я расскажу, как мы делали интеграционный слой для CRM-системы группы контакт-центров торговой сети Пятерочка.

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

На старте разработки контакт-центры Пятерочки располагались на 7 разных площадках. И все они использовали разные решения для учета звонков: платформы дискретных каналов, обработки обращений и т. д. Решения по обращениям велись в отдельной системе, а уведомления по обращениям делались в ручном режиме.

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

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

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

Возможные решения

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

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

Другой вариант enterprise service bus. Это классический сценарий интеграции: разработать пользовательский интерфейс, подключить его к опубликованным на интеграционной шине методам и наслаждаться жизнью.

Но ESB в 2020 году уже не самое популярное решение. К тому же, заказчику хотелось реализовать дополнительную функциональность, например, подсистему отчетности, а она требует разработки бэкенда.

Мы решили использовать для интеграционного слоя микросервисы. Они позволяют реализовать дополнительную бизнес-логику и обеспечивают масштабируемость. Их проще распределить по нескольким ЦОД, по сравнению с ESB.

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

Архитектура интеграционного слоя

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

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

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

Так как в интеграционный слой попала бизнес-логика, выбор пал на событийно-ориентированную архитектуру. В ней проще следить за согласованностью данных.

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

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

Вся сложность во взаимодействиях

С внешними системами было просто. Мы не могли ничего менять, так что использовали те API, которые они предоставляли.

Внутри системы мы выделили два типа взаимодействий: модифицирующие и читающие. Модифицирующие воздействия реализовали при помощи обмена сообщениями через топики в Kafka. Каждому сервису присвоили топик, который начинается с его имени. Сервис отправляет туда сообщения обо всех событиях, произошедших с его объектами:

с бизнес-сущностью, которую обрабатывает сервис user, что-то произошло, вот ее новые данные

Этот топик читают другие сервисы. Они реагируют на изменения и отписываются в свои топики. Так действие пользователя по каскаду распространяется по всей системе.

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

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

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

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

Kafka и очереди сообщений

Мы использовали Kafka в качестве брокера сообщений, потому что купились на отказоустойчивость. Между ЦОДами, где расположены системы заказчика, небольшие задержки, так что она работает неплохо. Однако, где-то в середине проекта мы поняли, что просчитались.

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

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

Взаимодействие с фронтом

Здесь обошлось без rocket science. Взаимодействие с фронтом мы выстроили при помощи WebSocket и REST. WebSocket нужен для передачи событий от бека к фронту и высоконагруженных управляющих взаимодействий, а REST для запроса данных. Правда, пришлось написать отдельный микросервис для фронтенд-разработчиков.

Ходить за несколькими сущностями в три разных сервиса, а потом собирать на фронте какую-то адекватную модель для представления не очень удобно. Куда удобнее использовать composer, который будет брать сразу несколько сущностей со связями между ними и компоновать в один JSON. Для этого отлично подошел Apache Camel. Правда, в следующий раз мы не станем заморачиваться с REST и используем graphQL.

Service mesh и безопасность

Service mesh нужен, чтобы не прописывать вручную всю логику межсервисных взаимодействий. Как правило, для соединения микросервисов мы используем Istio и в этом проекте тоже рассчитывали на эту платформу. Но в этом году все идет как-то не так, как ожидалось.

Мы столкнулись с такой версией OKD, на которую никак не получалось установить Istio. Заказчик собирался в будущем обновить систему оркестрации, но нам нужно было решение здесь и сейчас.

Дедлайн приближался, а инженеры не знали, когда мы справимся с проблемой. Пришлось минимизировать риски и сделать какое-то подобие service mesh, чтобы точно запуститься вовремя.

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

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

Итак, мы начали пилить собственный service mesh. В его основу легла простая идея: если нельзя (не хочется) поместить логику в код сервиса, сделаем внешний sidecar-контейнер по аналогии с Istio, использовав OpenResty и язык Lua.

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

Аудит

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

Существуют штатные библиотеки для работы с Kafka, но и с ними возникли проблемы. Все ломалось, как только для подключения к Kafka требовался SSL. Спас положение старый добрый kafka-console-producer, который лежит в каждом sidecar и вызывается с параметрами из nginx.

Получился рабочий почти-service mesh. Казалось бы, можно было расслабиться, но мы чуть не упустили еще одну важную вещь.

Балансировка gRPC

gRPC то работает на HTTP/2, а этот протокол устанавливает соединение и держит, пока оно не порвется. Возможна ситуация, когда инстанс сервиса A намертво прицепится к одному из инстансов сервиса B и будет направлять все запросы только на него. Тогда от масштабирования сервиса B не будет толка. Второй инстанс сервиса B останется не нагружен и будет впустую тратить процессорное время и память.

Мы могли сделать полноценную балансировку на клиентской стороне, но было бы неправильно тратить на это время. После внедрения полноценного service mesh эти наработки потеряют актуальность.

Поэтому мы реализовали перехват и подсчет всех вызовов gRPC на уровне кода на стороне клиента. Когда счетчики подходят к определенному значению, вызывается метод, приводящий к переустановке соединения, и балансировщик OKD задействует незагруженный инстанс сервиса B.

Конечно, не стоит повторять это идеальных условиях, но если возникли неожиданные препятствия, и нужно срочно запуститься, это решение имеет право на жизнь. Такая балансировка работает. Графики нагрузки по сервисам ровные.

Управление версиями

Напоследок, несколько слов о том, как мы управляли версиями кода.

В проекте задействованы Spring, Maven и Java. Каждый микросервис хранится в собственном микрорепозитории. В первую очередь потому что мы передавали систему заказчику по частям, сервис за сервисом.

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

Прежде всего, мы стали собирать сервисы в образы, которые тегируются коротким хешем коммита. А внутри Maven проекта нет версионирования кода, там вместо версии мы использовали заглушку.

Затем мы выделили отдельный репозиторий со списком всех сервисов и их версиями. Коммит в этот репозиторий и назывался релизом. То есть, мы формировали манифест совместимых между собой версий микросервисов. Конечно, не руками, а через самодельные веб-интерфейсы.

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

Если сервис A и сервис B общаются по gRPC, и разработчик сервиса B исправит что-то в протофайле и сломает обратную совместимость, это выстрелит только, когда выйдет в рантайм. Слишком поздно. Поэтому мы храним все интеграционные контракты (протофайлы, классы для сериализации Kafka и т. д.) отдельно и подключаем как саб-модули. Благодаря этому решению, многие возможные проблемы проявляются на этапе компиляции.

Что мы вынесли из проекта

В этом проекте мы вновь убедились в том, как важно заранее обследовать инфраструктуру. Но даже после тщательной проверки нельзя расслабляться. Некоторые проблемы, типа несовместимости конкретной версии OKD с Istio, практически невозможно предвидеть.

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

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

Если у вас появились вопросы, или вы знаете, как красиво сделать приоритизацию сообщений в Kafka, пишите в комментариях или на почту IBolshakov@croc.ru.

Pulsar или Kafka что лучше? Здесь мы обсудим плюсы и минусы, распространенные мифы и нетехнические критерии, чтобы найти лучший инструмент для ваших задач.

Обычно я рассказываю об Apache Kafka и ее экосистеме. О Pulsar за последние годы меня спрашивали только коммитеры и авторы Pulsar. Они задавали сложные технические вопросы, чтобы показать, что Kafka не идет ни в какое сравнение с Pulsar. На Reddit и подобных платформах разгораются яростные и очень субъективные споры на эту тему. Я поделюсь своей точкой зрения, основанной на многолетнем опыте работы со стриминговыми опенсорс-платформами.

Сравнение технологий нынче в моде: Kafka vs. Middleware, Event Streaming и API Platforms

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

При этом такие сравнения почти всегда субъективны. Даже если автор не работает на вендора и представляется независимым консультантом, скорее всего, у него все равно есть любимчики, даже если он сам себе в этом не признается. Однако полезно посмотреть на инструменты с разных точек зрения. Поскольку Apache Pulsar сейчас обсуждают, я решил поделиться своим мнением и сравнить его с Kafka. Я работаю в Confluent, а мы лучшие эксперты по Apache Kafka и ее экосистеме. И все же я постараюсь быть максимально объективным и оперировать голыми фактами.

Опенсорс-фреймворки и коммерческие программы постоянно сравнивают. Я и сам делал такие сравнения в своем блоге и на других платформах, например InfoQ: Сравнение платформ интеграции, Выбор подходящего ESB для ваших потребностей, Kafka vs. ETL / ESB / MQ, Kafka vs. Mainframe и Apache Kafka и API Management / API Gateway. Просто заказчики хотели понять, какой инструмент поможет решить те или иные задачи.

Если говорить о сравнении Pulsar и Kafka, ситуация немного отличается.

Зачем сравнивать Pulsar и Kafka?

Реальные и потенциальные заказчики редко спрашивают о Pulsar. Хотя, если честно, в последние месяцы чуть чаще где-то на каждой 15-й или 20-й встрече. Потому что фичи и варианты использования у них частично совпадают. Хотя, по-моему, все дело в паре статей о том, что Pulsar якобы в чем-то лучше Kafka. В пользу противоположной точки зрения нет никаких фактов и очень мало материала.

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

Например, их главный пользователь Tencent, крупная китайская технологическая компания, которая при этом очень активно использует Kafka везде, кроме одного проекта, где пригодился Pulsar. Tencent обрабатывает с Kafka десять триллионов сообщений в день (только представьте: 10 000 000 000 000). Если посчитать, Tencent использует Kafka в тысячу раз активнее, чем Pulsar (10 трлн против десятков млрд). Tencent с удовольствием рассказывают о своем деплое Kafka: Как Tencent PCG использует Apache Kafka для обработки 10+ трлн сообщений в день.

Сравнение двух конкурирующих опенсорс-платформ

Apache Kafka и Apache Pulsar две замечательные конкурирующие технологии. Логично будет их сравнить.

У Apache Kafka и Apache Pulsar очень схожий набор функций. Советую оценить обе платформы на предмет доступных функций, зрелости, распространения на рынке, опенсорс-инструментов и проектов, обучающего материала, проведения митапов по регионам, видео, статей и так далее. Примеры вариантов использования в вашей отрасли помогут принять правильное решение.

Confluent уже писал сравнение Kafka vs. Pulsar vs. RabbitMQ: производительность, архитектура и функции. Я тоже поучаствовал. Значит, сравнение уже есть

О чем тогда эта статья?

Я хочу рассмотреть несколько мифов из споров на тему Kafka против Pulsar, которые регулярно возникают в блогах и на форумах. Затем я сравню их не только по техническим аспектам, потому что обычно никто не говорит о Pulsar с этой стороны.

Kafka vs Pulsar мифы

Мне попадалось несколько мифов. С некоторыми я согласен, другие можно легко развенчать фактами. Ваше мнение может отличаться от моего, это нормально. Я излагаю исключительно свою точку зрения.

Миф 1. У Pulsar есть характеристики, которых нет у Kafka.

Правда.

Если сравнивать Apache Kafka и Apache Pulsar, можно найти различия в многоуровневой архитектуре, очередях и мультитенантности.

Но!

У Kafka тоже есть уникальные особенности:

• В два раза меньше серверов, которыми приходится управлять.
• Данные сохраняются на диск только один раз.
• Данные кэшируются в памяти только однажды.
• Проверенный протокол репликации.
• Производительность zero-copy.
• Транзакции.
• Встроенная обработка потока.
• Долгосрочное хранилище.
• В разработке: удаление ZooKeeper (KIP-500), чтобы использовать и деплоить Kafka было еще проще, чем Pulsar с его четырехкомпонентной архитектурой (Pulsar, ZooKeeper, BookKeeper и RocksDB), а еще чтобы повысить масштабируемость, устойчивость и т. д.
• В разработке: многоуровневое хранилище (KIP-405), чтобы повысить эластичность и экономичность Kafka.

Спросите себя: можно ли сравнивать опенсорс-платформы или продукты и вендоров с комплексным предложением?

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

Например. Понадобилось несколько лет, чтобы реализовать и испытать в бою Kafka Streams в качестве нативного движка обработки потоков. Как это можно сравнивать с Pulsar Functions? Последнее позволяет добавлять определяемые пользователем функции (UDF) безо всякой связи с обработкой реальных потоков. Или это скорее похоже на Single Message Transformations (SMT), основную функцию Kafka Connect? Не сравнивайте круглое с квадратным и не забывайте учитывать зрелость. Чем критичнее функция, тем больше зрелости ей требуется.

Сообщество Kafka вкладывает невероятные усилия в работу над основным проектом и его экосистемой. Только в Confluent 200 инженеров занимаются исключительно проектом Kafka, дополнительными компонентами, коммерческими продуктами и предложениями SaaS в основных облачных провайдерах.

Миф 2. У Pulsar есть несколько крупных пользователей, например китайский Tencent.

Правда.

Но! Tencent использует Kafka активнее, чем Pulsar. Отдел выставления счетов, где используется Pulsar, это лишь малая часть Tencent, в то время как остальные используют Kafka. У них есть глобальная архитектура на основе Kafka, где больше тысячи брокеров сгруппированы в единый логический кластер.

Будьте осторожны с опенсорс-проектами. Проверяйте, какого успеха добились с ними обычные компании. Если эту технологию использует один технологический гигант, это еще не значит, что вам она тоже подойдет. Сколько компаний из Fortune 2000 могут похвастаться историями успеха с Pulsar?

Ищите примеры не только среди гигантов!

Примеры обычных компаний позволят лучше понять, как применяется тот или иной инструмент в реальном мире. Если это не истории успеха от самих вендоров. На сайте Kafka можно найти много примеров компаний. Более того, на конференциях Kafka Summit в Сан-Франциско, Нью-Йорке и Лондоне каждый год разные предприятия из разных отраслей делятся своими историями и кейсами. Это компании из Fortune 2000, предприятия среднего бизнеса и стартапы.

Приведу один пример про Kafka. Для репликации данных в реальном времени между отдельными кластерами Kafka существует много разных инструментов, включая MirrorMaker 1 (часть проекта Apache Kafka), MirrorMaker 2 (часть проекта Apache Kafka), Confluent Replicator (от Confluent, доступен только в рамках Confluent Platform и Confluent Cloud), uReplicator (опенсорс от Uber), Mirus (опенсорс от Salesforce), Brooklin (опенсорс от LinkedIn).

На практике разумно использовать только два варианта, если, конечно, вы не хотите обслуживать и улучшать код самостоятельно. Это MirrorMaker 2 (еще не вполне зрелая новинка, но отличный выбор для средне- и долгосрочной перспективы) и Confluent Replicator (проверенный на практике во многих критически важных для бизнеса деплоях, но платный). Остальные варианты тоже нормальные. Но кто обслуживает эти проекты? Кто разбирается с багами и проблемами безопасности? Кому звонить, если в продакшене что-то случилось? Одно дело деплоить критически важные системы в продакшене, другое оценивать и пробовать опенсорс-проект.

Миф 3. Pulsar предлагает очереди и стриминг в одном решении.

Частично.

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

Pulsar предлагает только ограниченную поддержку очередей сообщений и стриминга событий. В обеих областях он пока не может составить крепкую конкуренцию по двум причинам:

1. Pulsar предлагает ограниченную поддержку очередей сообщений, потому что ему не хватает таких популярных функций, как XA-транзакции, маршрутизация, фильтрация сообщений и т. д. Это обычные функции в таких системах сообщений, как IBM MQ, RabbitMQ, и ActiveMQ. Адаптеры Pulsar для систем сообщений тоже ограничены в этом смысле. В теории все выглядит нормально, но на практике...

   
   

2. Pulsar предлагает ограниченную поддержку стриминга. Например, на практике в большинстве случаев он не поддерживает семантику строго однократной (exactly-once) доставки и обработки. Вряд ли кто-то станет использовать Pulsar для платежной системы, потому что платежи могут дублироваться и теряться. Ему не хватает функционала для обработки потоков с соединением, агрегированием, окнами, отказоустойчивым управлением состоянием и обработкой на основе времени событий. Топики в Pulsar отличаются от топиков в Kafka в худшую сторону из-за BookKeeper, который был придуман в 2008 году как лог упреждающей записи для HDFS namenode в Hadoop в расчете на краткосрочное хранение.

   
   

Примечание. Адаптер Kafka для Pulsar тоже ограничен. В теории звучит заманчиво, но в реальности не все получается, потому что он поддерживает только небольшую часть функционала Kafka.

Как и Pulsar, Kafka предлагает ограниченную поддержку очереди.

В Kafka можно использовать разные обходные пути, чтобы реализовать нормальное поведение очереди. Если вы хотите использовать отдельные очереди вместо общих топиков Kafka для:

• Безопасности => используйте Kafka ACL (и дополнительные инструменты, вроде контроля доступа на основе ролей (RBAC) от Confluent).
• Семантики (отдельных приложений) => используйте группы консюмеров в Kafka.
• Балансировки нагрузки => используйте партиции Kafka.

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

Зная обходные пути и ограничения Pulsar и Kafka в плане сообщений, давайте проясним: ни одна из платформ не предлагает решение для обмена сообщениями.

Если вам нужно именно оно, возьмите что-то вроде RabbitMQ или NATS.

Однозначного решения здесь нет. Многие наши заказчики заменяют имеющиеся системы сообщений, вроде IBM MQ, на Kafka (из соображений масштабируемости и затрат). Просто нужно знать имеющиеся варианты и их недостатки, все взвесить и подобрать решение, которое лучше всего подходит именно вам.

Миф 4. Pulsar предоставляет потоковую обработку.

Неправда.

Или, если по-честному, все зависит от того, что вы называете обработкой потоков. Это какая-то простейшая функция или прямо полноценная обработка?

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

Pulsar предлагает минимальную потоковую обработку через Pulsar Functions. Она подходит для простых обратных вызовов, но не сравнится с функционалом Kafka Streams или ksqlDB для создания стриминговых приложений со stateful-информацией, скользящими окнами и другими функциями. Варианты применения можно найти в разных отраслях. Например, на сайте Kafka Streams есть кейсы New York Times, Pinterest, Trivago, Zalando и других.

Для стриминга аналитики Pulsar обычно используется в сочетании с нормальной платформой обработки потоков, вроде Apache Spark или Apache Flink, но вам придется управлять дополнительным элементами распределенной архитектуры и разбираться в их сложном взаимодействии.

Миф 5. Pulsar предоставляет семантику exactly-once, как и Kafka.

Неправда.

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

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

Семантика Exactly-Once Semantics (EOS) доступна с версии Kafka 0.11, которая вышла три года назад, и используется во многих продакшен-деплоях. Kafka EOS поддерживает всю экосистему Kafka, включая Kafka Connect, Kafka Streams, ksqlDB и такие клиенты, как Java, C, C++, Go и Python. На конференции Kafka Summit было несколько выступлений, посвященных функционалу Kafka EOS, включая это превосходное и понятное введение со слайдами и видео.

Миф 6. У Pulsar производительность выше, чем у Kafka.

Неправда.

Я не поклонник разных бенчмарков производительности и пропускной способности. Обычно они субъективны и относятся к конкретной задаче независимо от того, кто их проводит вендор, независимый консультант или исследователь.

Например, GIGAOM опубликовали один бенчмарк, где сравнивается задержка и производительность в Kafka и Pulsar. Автор намеренно замедлил Kafka, настроив параметр flush.messages = 1, в результате которого каждый запрос вызывает fsync и Kafka синхронизируется с диском при каждом сообщении. В этом же бенчмарке консюмер Kafka отправляет подтверждение синхронно, а консюмер Pulsar асинхронно. Неудивительно, что Pulsar вышел явным победителем. Правда, автор забыл упомянуть или объяснить существенные отличия в конфигурации и измерениях. Давайте не будем путать теплое с мягким.

На самом деле архитектура Pulsar сильнее нагружает сеть (из-за брокера, который выступает как прокси перед серверами bookie BookKeeper) и требует в два раза больше пропускной способности (потому что BookKeeper записывает данные в лог упреждающей записи и в главный сегмент).

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

Оцените свои требования к производительности. Сделайте proof-of-concept с Kafka и Pulsar, если надо. Скорее всего, в 99% случаев обе платформы покажут приемлемую производительность для вашего сценария. Не доверяйте посторонним субъективным бенчмаркам! Ваш случай все равно уникален, и производительность это только один из аспектов.

Миф 7. Pulsar проще в использовании, чем Kafka.

Неправда.

Без дополнительных инструментов вам будет сложно и с Kafka, и с Pulsar.

У Kafka две распределенных системы: сама Kafka и Apache ZooKeeper.

Но! У Pulsar три распределенных системы, а еще хранилище: Pulsar, ZooKeeper и Apache BookKeeper. Как и Pulsar, BookKeeper использует ZooKeeper. Для некоторых задач хранения используется RocksDB. Это означает, что Pulsar гораздо сложнее понять и настроить по сравнению с Kafka. Кроме того, у Pulsar больше параметров конфигурации, чем у Kafka.

Kafka движется в противоположном направлении скоро ZooKeeper будет удален (см. KIP-500), так что останется всего одна распределенная система, которую нужно деплоить, обслуживать, масштабировать и мониторить:

ZooKeeper мешает масштабированию в Kafka и усложняет эксплуатацию. Но у Pulsar все еще хуже!

Одна из главных проблем моих заказчиков использование ZooKeeper в критически важных деплоях в большом масштабе. С нетерпением жду упрощения архитектуры Kafka, чтобы остались только брокеры. Кроме того, мы получим единую модель безопасности, потому что больше не придется отдельно настраивать безопасность для ZooKeeper. Это огромное преимущество, особенно для крупных организаций и отраслей со строгим регулированием. Ребята из комплаенса и ИБ скажут вам спасибо за упрощенную архитектуру.

Использование платформы связано НЕ только с архитектурой

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

Кроме того, по Kafka есть много вариантов обучения онлайн-курсы, книги, митапы и конференции. К сожалению, у Pulsar с этим все плохо.

Миф 8. Архитектура с тремя уровнями лучше, чем с двумя.

Зависит от ситуации.

Лично я скептически отношусь к тому, что трехуровневую архитектуру Pulsar (брокеры Pulsar, ZooKeeper и BookKeeper) можно считать преимуществом для большинства проектов. Тут есть издержки.

Twitter рассказал, как отказался от BookKeeper + DistributedLog (DistributedLog похож на Pulsar по архитектуре и дизайну) около года назад, соблазнившись такими преимуществами одноуровневой архитектуры Kafka, как экономичность и улучшенная производительность, которых недоставало двухуровневой архитектуре с отдельным хранилищем.

Как и Pulsar, DistributedLog построен на BookKeeper и добавляет стриминговый функционал, схожий с Pulsar (например, уровень обработки существует отдельно от уровня хранилища). Изначально DistributedLog был отдельным проектом, но потом присоединился к BookKeeper, хотя сегодня им, похоже, мало кто занимается (всего пара коммитов за последний год). Среди основных причин перехода Twitter на Kafka называлась значительная экономия и повышение производительности, а также обширное сообщество и популярность Kafka. Вот какие выводы они сделали: Для одного консюмера экономия ресурсов составила 68%, а для нескольких целых 75%.

Преимущества трехуровневой архитектуры проявляются при создании масштабируемой инфраструктуры. При этом дополнительный уровень увеличивает нагрузку на сеть минимум на 33%, а данные, которые хранятся в брокерах Pulsar, приходится дополнительно кэшировать на обоих уровнях, чтобы добиться аналогичной производительности, а еще дважды записывать на диск, потому что формат хранения в Bookkeeper основан не на логе.

В облаке, где работает большинство деплоев Kafka, лучшим внешним хранилищем выступает не такая узкоспециализированная технология, как BookKeeper, а популярное и проверенное хранилище объектов, вроде AWS S3 и GCP GCS.

Tiered Storage в Confluent Platform на основе AWS S3, GCP GCS и им подобных, дает те же преимущества без дополнительного слоя BookKeeper, как у Pulsar, и без дополнительных затрат и задержек, связанных с передачей данных по сети. У Confluent ушло два года на то, чтобы выпустить GA-версию Tiered Storage for Kafka с круглосуточной поддержкой для самых критичных данных. Tiered Storage пока недоступно для опенсорсной версии Apache Kafka, но Confluent вместе с сообществом Kafka (включая крупные технологические компании, вроде Uber) работает над KIP-405, чтобы добавить Tiered Storage в Kafka с разными вариантами хранения.

У обеих архитектур есть плюсы и минусы. Лично мне кажется, что 95% проектов не нуждаются в сложной трехуровневой архитектуре. Она может пригодиться там, где требуется внешнее недорогое хранилище, но вам придется отвечать за круглосуточное соблюдение SLA, масштабируемость и пропускную способность. А еще за безопасность, управление, поддержку и интеграцию в вашу экосистему. Если вам действительно нужна трехуровневая архитектура, не стану вас останавливать.

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

Неправда.

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

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

Миф 9. Kafka масштабируется хуже, чем Pulsar.

Неправда.

Это один из главных аргументов сообщества Pulsar. Как я уже говорил, это всегда зависит от выбранного бенчмарка. Например, я видел тесты с эквивалентными вычислительными ресурсами, где на высокой пропускной способности у Kafka результат был гораздо лучше, чем у Pulsar. Вот один из таких бенчмарков:

В большинстве случаев масштабируемость не проблема. Kafka можно легко разогнать до нескольких гигабайтов в секунду, как в демонстрации Масштабирование Apache Kafka до 10+ ГБ в секунду в Confluent Cloud:

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

Связанный миф: сейчас в Kafka может храниться всего 500к партиций на кластер.

Правда.

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

Но! Pulsar тоже не резиновый. Просто у него другие лимиты.

Ограничения в Kafka связаны с Zookeeper. Когда Zookeeper удалят через KIP-500, верхняя граница уйдет вместе с ним.

Примечание: успех зависит от подходящего устройства архитектуры!

Если у заказчика возникли проблемы с количеством партиций и масштабируемостью в Kafka, скорее всего, он ошибся в архитектуре и приложениях (Pulsar бы тут точно не помог).

Kafka это платформа стриминга, а не очередной IBM MQ. Если вы попробуете воссоздать любимое решение и архитектуру MQ с Kafka, у вас вряд ли что-то получится. У меня было несколько заказчиков, которые потерпели крах, но смогли привести все в порядок с нашей помощью.

Скорее всего, у вас не возникнет проблем с количеством партиций и масштабируемостью, даже пока Kafka еще использует ZooKeeper, если вы правильно все спроектируете и разберетесь в базовых принципах Kafka. Такие проблемы часто возникают с любой технологией, вроде Kafka, которая вводит совершенно новый уровень и парадигму (например, при первом появлении облаков немало компаний набили себе шишки при попытке их освоить).

Связанный миф: Pulsar поддерживает практически неограниченное число партиций.

Неправда.

У BookKeeper существуют те же ограничения по одному файлу на ledger, что и у Kafka, но в одной партиции таких ledger несколько. Брокеры Pulsar объединяют партиции в группы, но на уровне хранения, в Bookkeeper, данные хранятся в сегментах, и на каждую партицию приходится много сегментов.

В Kafka метаданные для этих сегментов хранятся в Zookeeper, откуда и возникают эти ограничения по числу. Когда Kafka избавится от этой зависимости, границы возможного раздвинутся. С нетерпением жду реализации KIP-500. Подробности читайте в статье Apache Kafka справится сама: удаление зависимости от Apache ZooKeeper.

Связанный миф: потребности масштабирования в Kafka необходимо определять при создании топика.

Отчасти это правда.

Если требуется очень большой масштаб, в топиках Kafka можно сразу создать больше партиций, чем обычно требуется для этой задачи. (См. Потоки и таблицы в Apache Kafka: топики, партиции и хранение.) Или можно добавить партиции позже. Это не идеальное решение, но так уж устроены распределенные системы стриминга (которые, кстати, масштабируются лучше, чем традиционные системы обработки сообщений, вроде IBM MQ).

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

Но! У топиков Pulsar есть то же ограничение!

Пропускная способность для записи зависит от числа партиций, назначенных топику Pulsar, точно так же, как это происходит в топике Kafka, поэтому там тоже придется создавать партиции с запасом. Дело в том, что в каждой партиции записывать можно только в один ledger за раз (а их может быть несколько). Увеличение числа партиций динамически влияет на порядок сообщений, как и в Kafka (порядок нарушается).

У Kafka и Pulsar с масштабированием все отлично. Этого будет более чем достаточно почти в любом сценарии.

Если вам нужен совсем заоблачный масштаб, лучше взять реализацию без ZooKeeper. Так что KIP-500 это самое ожидаемое изменение в Kafka, судя по сообществу и заказчикам Confluent.

Миф 10. В случае аппаратного сбоя Pulsar восстанавливается моментально, а Kafka приходится перегружать данные.

И да, и нет.

Если брокер Pulsar вырубится, ничего не страшного не будет, это правда, но, в отличие от брокера Kafka, он и не хранит данные, а просто выступает в качестве прокси для уровня хранения, то есть BookKeeper. Так что подобные заявления о Pulsar это просто маркетинговый ход. Почему никто не говорит, что бывает, если полетит нода BookKeeper (bookie)?

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

Kafka уже предлагает эластичность.

Это важно. Основатель Confluent, Джей Крепс (Jay Kreps) недавно писал об этом: Эластичные кластеры Apache Kafka в Confluent Cloud. В облачном сервисе SaaS, вроде Confluent Cloud, конечный пользователь не думает об аппаратных сбоях. Он ожидает непрерывный аптайм и SLA на уровне 99,xx. С оплатой за потребление пользователь не заботится об управлении брокерами, изменении размеров нод, увеличении или уменьшении кластеров и подобных деталях.

Самоуправляемым кластерам Kafka нужны такие же возможности. Tiered Storage for Kafka это огромное хранилище, с которым данные не хранятся на брокере и восстановление после сбоев происходит почти моментально. Если добавить сюда такие инструменты, как Self-Balancing Kafka (эта фича от Confluent описана в статье по ссылке выше), можно вообще забыть об эластичности в самоуправляемых кластерах.

К сожалению, в Pulsar вы ничего подобного не найдете.

Миф 11. Pulsar справляется с репликацией между кластерами лучше, чем Kafka.

Неправда.

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

Задачи по кворуму Kafka поручает ZooKeeper. Даже после реализации KIP-500 и удаления ZooKeeper законы физики не перестанут действовать: проблемы задержки в распределенных системах существуют в таких регионах, как Восточная, Центральная и Западная часть США и даже по всему миру. Скорость света, конечно, впечатляет, но все же имеет ограничения.

Эту проблему можно обойти разными способами, включая использование инструментов репликации в реальном времени, например MirrorMaker 2 в Apache Kafka, Confluent Replicator или Confluent Multi-Region-Clusters. Эти варианты и советы см. в статье Архитектурные шаблоны для распределенных, гибридных, периферийных и глобальных деплоев Apache Kafka.

Вы не найдете универсальное решение, которое обеспечит глобальную репликацию + нулевой простой + нулевую потерю данных. Для самых критически важных приложений Confluent Multi-Region-Clusters предлагает RTO=0 и RPO=0 (нулевой простой и нулевую потерю данных) с автоматическим аварийным восстановлением и отработкой отказа, даже если упадет целый датацентр или облачный регион.

Для таких случаев архитектура Pulsar будет еще сложнее, чем ее базовая версия. Просто для георепликации Pulsar требует дополнительный глобальный кластер Zookeeper, а для больших расстояний это не годится. Обходной путь есть, но от CAP-теоремы и законов физики не уйдешь.

Что с Kafka, что с Pulsar вам нужна проверенная архитектура для борьбы с физикой в глобальных деплоях.

Миф 12. Pulsar совместим с интерфейсом и API Kafka.

Отчасти.

Pulsar предлагает простейшую реализацию с зачаточной совместимостью с протоколом Kafka v2.0.

У Pulsar есть конвертер для базовых элементов протокола Kafka.

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

Мы слышали заявления о совместимости с Kafka и в других примерах, например для гораздо более зрелого сервиса Azure Event Hubs. Почитайте об ограничивающих факторах их Kafka API. Вы удивитесь. Никакой поддержки основных функций Kafka, вроде транзакций (и семантики exactly-once), сжатия или compaction для логов.

Раз, по сути, это не Kafka, существующие приложения Kafka будут вести себя непредсказуемо в таких вот совместимых системах, будь то Azure Event Hubs, Pulsar или другая оболочка.

Kafka vs. Pulsar комплексное сравнение

Выше мы говорили о мифах, которые гуляют по разным статьям и постам. С ними, вроде, все понятно.

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

Рассмотрим три из них: доля на рынке, корпоративная поддержка и облачные предложения.

Доля на рынке у Apache Kafka и Apache Pulsar

Статистика в Google Trends за последние пять лет совпадает с моими личными наблюдениями интерес к Apache Kafka гораздо выше, чем к Apache Pulsar:

Примерно так же выглядит картинка сравнения на Stack Overflow и аналогичных платформах, по числу и размеру поддерживающих вендоров, открытой экосистеме (интеграция инструментов, обертки, вроде Spring Kafka) и подобным характеристикам.

Открытые вакансии еще один индикатор распространения технологии. Для Pulsar их мало, то есть его использует не так много компаний. Убедитесь сами поищите на любом сайте. Если искать по всему миру, для Pulsar вакансий меньше сотни, а для Kafka тысячи. Кроме того, в большинстве вакансий для Pulsar указано, что требуется опыт с Kafka, Pulsar, Kinesis и аналогичными технологиями.

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

Раз Pulsar так мало распространен, почему о нем вообще говорят? Во-первых, потому, что независимые консультанты, аналитики и блоггеры (включая меня) должны рассказывать о новых современных технологиях, чтобы привлечь аудиторию. Если честно, люди любят об этом читать.

Корпоративная поддержка для Kafka и Pulsar

Корпоративная поддержка для Kafka и Pulsar существует.

Но все не совсем так, как можно ожидать. Вот несколько вендоров, к которым вы можете обратиться, если решили работать с Pulsar:

• Streamlio (теперь принадлежит Splunk) компания, которая раньше стояла за Apache Pulsar. Splunk пока не объявили о своей будущей стратегии для поддержки пользователей, которые работают над проектами на основе Pulsar. Splunk славится своими популярными аналитическими платформами. Это их основной бизнес (1,8 млрд долларов в 2019 году). Единственное, на что жалуются пользователи, ценник. Splunk активно используют Kafka, а сейчас интегрируют Pulsar в Splunk Data Stream Processor (DSP). Сильно сомневаюсь, что Splunk вдруг увлечется опенсорсом, чтобы поддержать ваш драгоценный проект на базе Pulsar (возможно, чего-то можно ожидать от DSP). Будущее покажет.
• StreamNative основана одним из изначальных разработчиков Apache Pulsar и поддерживает платформу стриминга на основе Pulsar. На июнь 2020 года у StreamNative было 13 (целых 13, да!) сотрудников в LinkedIn. Уж не знаю, хватит этого для поддержки вашего критически важного проекта или нет.
• TIBCO объявили о поддержке Pulsar в декабре 2019 года. За последние годы стратегия у них сместилась с интеграции на аналитику. Пользователи их промежуточного ПО уходят от TIBCO толпами. В отчаянии они приняли стратегическое решение: поддерживать другие платформы, даже если у них нет никакого опыта и никакой связи с ними. Да, звучит как миф, но TIBCO делают то же самое для Kafka. Любопытный факт: TIBCO предлагает Kafka и ZooKeeper в Windows! Такого больше никто не делает. Все знают, это приводит к нестабильности и несогласованности. Но если что, TIBCO помогут вам с Kafka и Pulsar. Зачем вообще сравнивать эти платформы, если можно получить обе у одного вендора, причем даже на Windows, с .exe и bat-скриптами для запуска серверных компонентов:

Вендоров для Kafka больше с каждым днем.

Это очень распространенная на рынке платформа. Лучшее подтверждение поддержка от крупнейших вендоров. IBM, Oracle, Amazon, Microsoft и многие другие поддерживают Kafka и встраивают возможности интеграции с ней в свои продукты.

Окончательное подтверждение этому я получил на конференции Oracle OpenWorld 2019 в Сан-Франциско, где выступал менеджер по GoldenGate (великолепный и очень дорогой инструмент CDC от Oracle). В основном он рассказывал, как GoldenGate станет платформой интеграции данных вообще для всего. Половина выступления была посвящена стримингу, платформе Kafka и тому, что GoldenGate будет обеспечивать интеграцию с разными базами/озерами данных и Kafka в обоих направлениях.

Полностью управляемые облачные сервисы для Kafka и Pulsar

Какие облачные решения доступны для Kafka и Pulsar?

Для Apache Pulsar есть облачный сервис с очень говорящим названием:

Kafkaesque.

Я серьезно. Сами посмотрите [UPD: 17 июня они переименовали KAFKAESQUE в KESQUE. Видимо, поняли, всю комичность ситуации.]

Для Apache Kafka существуют разные предложения на выбор:

• Confluent Cloud (SaaS) полностью управляемый сервис с оплатой по мере использования, круглосуточной доступностью, эластичностью и бессерверными функциями для Apache Kafka и связанной экосистемы (Schema Registry, коннекторы Kafka Connect и ksqlDB для обработки потоков).
• Amazon MSK (PaaS) подготавливает ZooKeeper и брокеры Kafka, а конечные пользователи сами обслуживают их, исправляют баги, устанавливают обновления и т. д. Важный факт: AWS не включает проблемы с Kafka в поддержку и SLA 99,95.
• Azure Event Hubs (SaaS) предоставляет конечную точку Kafka (с проприетарной реализацией) для взаимодействия с приложениями Kafka. Это легко масштабируемое решение с высокой производительностью. Это не вполне Kafka, просто эмуляция, и тут не хватает некоторых важных функций, вроде семантики exactly-once, compaction логов и сжатия. Не говоря уже о дополнительных возможностях, вроде Kafka Connect и Kafka Streams
• Big Blue (IBM) и Big Red (Oracle) предлагают облачные решения для Kafka и соответствующих API. Не знаю, использует их кто-нибудь или нет. Понятия не имею, насколько они хороши, никогда не видел их в деле.
• Много компаний поменьше, вроде Aiven, CloudKarafka, Instaclustr и других.

В общем, про распространенность Kafka и Pulsar на рынке все очевидно.

И все же Apache Kafka или Apache Pulsar?

В двух словах: Pulsar пока здорово отстает от Kafka по зрелости в плане надежности для больших масштабов и обширности сообщества.

И вообще не факт, что Pulsar лучше.

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

Если в Kafka и Pulsar есть не все нужные вам функции, выбирайте Kafka: для нее есть много вендоров и облачных решений. Pulsar, к сожалению, сильно отстает и вряд ли догонит в ближайшем будущем.

В основе Apache Kafka находится лог простая структура данных, которая использует последовательные операции, работающие в симбиозе с оборудованием. Эффективное использование дискового буфера и кэша процессора, prefetch, передача данных zero-copy и много других радостей все это благодаря построенной на логе структуре, которая славится своей эффективностью и пропускной способностью. Обычно эти преимущества, а еще базовая реализация в виде лога коммитов, первое, что люди узнают о Kafka.

Код самого лога составляет относительно малую часть всей системы. Гораздо больше занимает код, который отвечает за организацию партиций (т. е. логов) на множестве брокеров в кластере назначает лидеров, обрабатывает сбои и т. д. Этот код и делает Kafka надежной распределенной системой.

Раньше важной частью работы распределенного кода был Apache ZooKeeper. Он хранил самые важные метаданные системы: где находятся партиции, кто из реплик лидер и т. д. Поначалу эффективный и проверенный ZooKeeper действительно был нужен, но, по сути, это что-то вроде специфической файловой системы или API триггеров поверх согласованного лога. А Kafka это API pub/sub (издатель-подписчик) поверх согласованного лога. В результате пользователи настраивают, конфигурируют, мониторят, защищают и обдумывают взаимодействие и производительность между двумя реализациями лога, двумя сетевыми уровнями и двумя реализациями системы безопасности каждая со своими инструментами и хуками мониторинга. Все стало неоправданно сложно, поэтому решено было заменить ZooKeeper сервисом кворума прямо в самой Kafka.

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

Мы с Джейсоном, Колином и командой KIP-500 прогнали полный жизненный цикл сервера Kafka с созданием и потреблением сообщений и все без Zookeeper. Какая же красота!
Бен Стопфорд (Ben Stopford)

Рады сообщить, что ранний доступ к KIP-500 уже закоммичен в ствол и будет включен в предстоящий релиз 2.8. Впервые Kafka можно использовать без ZooKeeper. Мы называем это режимом Kafka Raft Metadata, сокращенно KRaft (читается крафт).

В этом релизе пока доступны не все фичи. Мы еще не поддерживаем ACL и другие транзакции или функции безопасности. Переназначение партиций и JBOD тоже не поддерживаются в режиме KRaft (надеемся, что они будут доступны в одном из релизов Apache Kafka в этом году). Так что контроллер кворума считаем экспериментальным и для прода пока не используем. В остальном преимуществ масса: деплоймент и обслуживание станут проще, всю Kafka можно запустить как один процесс, а на кластер помещается гораздо больше партиций (цифры см. ниже).

Контроллер кворума: консенсус на основе событий

Если запустить Kafka с новым контроллеров кворума, все задачи по метаданным, которые раньше выполнялись контроллером Kafka и ZooKeeper, ложатся на новый сервис, запущенный прямо в кластере Kafka. Контроллер кворума можно запустить на выделенном оборудовании, если этого требует ваш сценарий.

Внутри происходит много интересного. Контроллеры кворума используют новый протокол KRaft, чтобы точно реплицировать метаданные по всему кворуму. Протокол во многом напоминает ZooKeeper ZAB и Raft, но с важными отличиями например, что логично, он использует архитектуру на основе событий.

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

Консенсус на основе событий

Раз протокол KRaft управляется событиями, в отличие от контроллера ZooKeeper контроллер кворума не загружает стейт из ZooKeeper, когда становится активным. При смене лидера у нового активного контроллера уже есть в памяти все закоммиченные записи метаданных. Более того, для отслеживания метаданных в кластере используется тот же механизм на основе событий, что и в протоколе KRaft. Задача, которую раньше обрабатывали RPC, теперь полагается на события и использует лог для коммуникации. Приятное последствие этих изменений (и изменений, которые были предусмотрены изначальным замыслом) Kafka теперь поддерживает гораздо больше партиций, чем раньше. Обсудим подробнее.

Увеличение масштаба Kafka до миллионов партиций

Максимальное число партиций в кластере Kafka определяется двумя свойствами: лимит партиций для ноды и лимит партиций для кластера. Сейчас на ограничение в кластере во многом влияет управление метаданными. Предыдущие KIP улучшили лимит для ноды, хотя и здесь нет предела совершенству. Масштабируемость в Kafka зависит от добавления нод. Здесь нам важен лимит для кластера, который определяет верхнюю границу масштабирования всей системы.

Новый контроллер кворума может обрабатывать гораздо больше партиций на кластере. Чтобы это проверить, можно провести такие же тесты, как в 2018 году, с помощью которых мы демонстрировали лимиты партиций в Kafka. Эти тесты измеряют время, которое требуется для завершения работы и восстановления. В старом контроллере это операция O (число партиций). Она устанавливает верхнюю границу числа партиций, которые сейчас поддерживает один кластер Kafka.

Предыдущая реализация, как рассказывалось в посте по ссылке, могла достигать 200K партиций. Ограничивающим фактором служило время, которое требовалось для переноса важных метаданных между внешним консенсусом (ZooKeeper) и внутренним управлением лидером (контроллер Kafka). С новым контроллером кворума обе роли выполняет один компонент. Подход на основе событий означает, что контроллер отрабатывает отказ почти моментально. Ниже приводятся цифры нашего тестирования для кластера с 2 млн партиций (в 10 раз больше предыдущего лимита):

Важно учитывать контролируемую и неконтролируемую остановку. Контролируемые остановки включают типичные рабочие сценарии, вроде rolling restart стандартная процедура развертывания изменений без потери доступности. Восстановление после неконтролируемой остановки важнее, потому что оно задает целевое время восстановления (RTO) системы в случае, скажем, неожиданного сбоя, например, на виртуальной машине или поде, или потери связи с дата-центром. Хотя эти цифры указывают на производительность всей системы, они напрямую определяют хорошо известное узкое место, которое возникает при использовании ZooKeeper.

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

Упрощение Kafka единый процесс

Kafka часто воспринимается как тяжеловес, и такой репутацией она во многом обязана сложностями с управлением ZooKeeper. Из-за этого для запуска проектов часто выбирают более легкую очередь сообщений, вроде ActiveMQ или RabbitMQ, а Kafka появляется на сцене только при разрастании масштаба.

Это печально, ведь Kafka дает удобную абстракцию на основе лога коммитов, которая одинаково хорошо подходит для маленьких рабочих нагрузок у стартапов и для бешеного трафика у Netflix или Instagram. А если вам нужна стриминговая обработка, без Kafka с абстракцией лога коммитов не обойтись, будь то Kafka Streams, ksqlDB или другой аналогичный фреймворк. Управлять двумя отдельными системами (Kafka и Zookeeper) сложно, поэтому пользователи выбирали между масштабированием и простотой.

Теперь можно получить все и сразу. Благодаря KIP-500 и режиму KRaft мы можем легко начать проект с Kafka или использовать его как альтернативу монолитным брокерам, вроде ActiveMQ или RabbitMQ. Благодаря легкости и единому процессу это подходящий вариант для ограниченных в ресурсах устройств. В облаке все еще проще. Управляемые сервисы, например, Confluent Cloud, все берут на себя. Не важно, собственный это будет кластер или управляемый, мы можем начать с малого, а потом расширять его вместе со своими рабочими нагрузками все в рамках одной инфраструктуры. Давайте посмотрим, как выглядит развертывание с одним процессом.

Испытываем Kafka без ZooKeeper

Новый контроллер кворума пока доступен в экспериментальном режиме, но, скорее всего, будет включен в релиз Apache Kafka 2.8. На что он способен? Самое крутое это, конечно, возможность создавать кластер Kafka с одним процессом, как в демо.

Конечно, если вам нужна очень высокая пропускная способность, и вы хотите добавить репликацию для отказоустойчивости, понадобятся еще процессы. Пока контроллер кворума на базе KRaft доступен только как Early Access, для критических рабочих нагрузок он не годится. В ближайшие месяцы мы будем добавлять недостающие фрагменты, моделировать протокол с помощью TLA+ и внедрять контроллер кворума в Confluent Cloud.

Но поэкспериментировать вы можете уже сейчас. См. полный README на GitHub.

Гостем подкаста The Art Of Programming стал спикер курса Слёрма по Kafka Александр Миронов, Infrastructure Engineer в Stripe. Тема выпуска Проблемная Kafka. Обсудили вопросы, часто возникающие при работе с Kafka: аудит входных данных, квоты, способы хранения данных, возможный даунтайм в консьюмер-группах и др.

Итоговый проект по Kafka

Релиз видеокурса от Александра Миронова и Анатолия Солдатова состоялся 7 апреля. Кроме заявленных тем мы добавили в курс итоговый проект: студентам нужно будет выполнить практическое задание под звёздочкой с применением знаний, полученных на курсе.

Мы добавили два варианта заданий из области разработки и из области администрирования. В первом варианте понадобится создать приложение для генерации real-time статистики продаж интернет-магазина, а во втором спасти кластер от хаоса.

Подробнее о курсе
Бесплатные материалы курса

Введение

На текущий момент не так много примеров тестов для приложений на основе Spark Structured Streaming. Поэтому в данной статье приводятся базовые примеры тестов с подробным описанием.

Все примеры используют: Apache Spark 3.0.1.

Подготовка

Необходимо установить:

• Apache Spark 3.0.x
• Python 3.7 и виртуальное окружение для него
• Conda 4.y
• scikit-learn 0.22.z
• Maven 3.v
• В примерах для Scala используется версия 2.12.10.

1. Загрузить Apache Spark
2. Распаковать: tar -xvzf ./spark-3.0.1-bin-hadoop2.7.tgz
3. Создать окружение, к примеру, с помощью conda: conda create -n sp python=3.7

Необходимо настроить переменные среды. Здесь приведен пример для локального запуска.

SPARK_HOME=/Users/$USER/Documents/spark/spark-3.0.1-bin-hadoop2.7PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9-src.zip;

Тесты

Пример с scikit-learn

При написании тестов необходимо разделять код таким образом, чтобы можно было изолировать логику и реальное применение конечного API. Хороший пример изоляции: DataFrame-pandas, DataFrame-spark.

Для написания тестов будет использоваться следующий пример: LinearRegression.

Итак, пусть код для тестирования использует следующий "шаблон" для Python:

class XService:    def __init__(self):        # Инициализация    def train(self, ds):        # Обучение    def predict(self, ds):        # Предсказание и вывод результатов

Для Scala шаблон выглядит соответственно.

Полный пример:

from sklearn import linear_modelclass LocalService:    def __init__(self):        self.model = linear_model.LinearRegression()    def train(self, ds):        X, y = ds        self.model.fit(X, y)    def predict(self, ds):        r = self.model.predict(ds)        print(r)

Тест.

Импорт:

import unittestimport numpy as np

Основной класс:

class RunTest(unittest.TestCase):

Запуск тестов:

if __name__ == "__main__":    unittest.main()

Подготовка данных:

X = np.array([    [1, 1],  # 6    [1, 2],  # 8    [2, 2],  # 9    [2, 3]  # 11])y = np.dot(X, np.array([1, 2])) + 3  # [ 6  8  9 11], y = 1 * x_0 + 2 * x_1 + 3

Создание модели и обучение:

service = local_service.LocalService()service.train((X, y))

Получение результатов:

service.predict(np.array([[3, 5]]))service.predict(np.array([[4, 6]]))

Ответ:

[16.][19.]

Все вместе:

import unittestimport numpy as npfrom spark_streaming_pp import local_serviceclass RunTest(unittest.TestCase):    def test_run(self):        # Prepare data.        X = np.array([            [1, 1],  # 6            [1, 2],  # 8            [2, 2],  # 9            [2, 3]  # 11        ])        y = np.dot(X, np.array([1, 2])) + 3  # [ 6  8  9 11], y = 1 * x_0 + 2 * x_1 + 3        # Create model and train.        service = local_service.LocalService()        service.train((X, y))        # Predict and results.        service.predict(np.array([[3, 5]]))        service.predict(np.array([[4, 6]]))        # [16.]        # [19.]if __name__ == "__main__":    unittest.main()

Пример с Spark и Python

Будет использован аналогичный алгоритм LinearRegression. Нужно отметить, что Structured Streaming основан на тех же DataFrame-х, которые используются и в Spark Sql. Но как обычно есть нюансы.

Инициализация:

self.service = LinearRegression(maxIter=10, regParam=0.01)self.model = None

Обучение:

self.model = self.service.fit(ds)

Получение результатов:

transformed_ds = self.model.transform(ds)q = transformed_ds.select("label", "prediction").writeStream.format("console").start()return q

Все вместе:

from pyspark.ml.regression import LinearRegressionclass StructuredStreamingService:    def __init__(self):        self.service = LinearRegression(maxIter=10, regParam=0.01)        self.model = None    def train(self, ds):        self.model = self.service.fit(ds)    def predict(self, ds):        transformed_ds = self.model.transform(ds)        q = transformed_ds.select("label", "prediction").writeStream.format("console").start()        return q

Сам тест.

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

train_ds = spark.createDataFrame([    (6.0, Vectors.dense([1.0, 1.0])),    (8.0, Vectors.dense([1.0, 2.0])),    (9.0, Vectors.dense([2.0, 2.0])),    (11.0, Vectors.dense([2.0, 3.0]))],    ["label", "features"])

Это очень удобно и код получается компактным.

Но подобный код, к сожалению, не будет работать в Structured Streaming, т.к. созданный DataFrame не будет обладать нужными свойствами, хотя и будет соответствовать контракту DataFrame.
На текущий момент для создания источников для тестов можно использовать такой же подход, что и в тестах для Spark.

def test_stream_read_options_overwrite(self):    bad_schema = StructType([StructField("test", IntegerType(), False)])    schema = StructType([StructField("data", StringType(), False)])    df = self.spark.readStream.format('csv').option('path', 'python/test_support/sql/fake') \        .schema(bad_schema)\        .load(path='python/test_support/sql/streaming', schema=schema, format='text')    self.assertTrue(df.isStreaming)    self.assertEqual(df.schema.simpleString(), "struct<data:string>")

И так.

Создается контекст для работы:

spark = SparkSession.builder.enableHiveSupport().getOrCreate()spark.sparkContext.setLogLevel("ERROR")

Подготовка данных для обучения (можно сделать обычным способом):

train_ds = spark.createDataFrame([    (6.0, Vectors.dense([1.0, 1.0])),    (8.0, Vectors.dense([1.0, 2.0])),    (9.0, Vectors.dense([2.0, 2.0])),    (11.0, Vectors.dense([2.0, 3.0]))],    ["label", "features"])

Обучение:

service = structure_streaming_service.StructuredStreamingService()service.train(train_ds)

Получение результатов. Для начала считываем данные из файла и выделяем: признаки и идентификатор для объектов. После запускаем предсказание с ожиданием в 3 секунды.

def extract_features(x):    values = x.split(",")    features_ = []    for i in values[1:]:        features_.append(float(i))    features = Vectors.dense(features_)    return featuresextract_features_udf = udf(extract_features, VectorUDT())def extract_label(x):    values = x.split(",")    label = float(values[0])    return labelextract_label_udf = udf(extract_label, FloatType())predict_ds = spark.readStream.format("text").option("path", "data/structured_streaming").load() \    .withColumn("features", extract_features_udf(col("value"))) \    .withColumn("label", extract_label_udf(col("value")))service.predict(predict_ds).awaitTermination(3)

Ответ:

15.9669918.96138

Все вместе:

import unittestimport warningsfrom pyspark.sql import SparkSessionfrom pyspark.sql.functions import col, udffrom pyspark.sql.types import FloatTypefrom pyspark.ml.linalg import Vectors, VectorUDTfrom spark_streaming_pp import structure_streaming_serviceclass RunTest(unittest.TestCase):    def test_run(self):        spark = SparkSession.builder.enableHiveSupport().getOrCreate()        spark.sparkContext.setLogLevel("ERROR")        # Prepare data.        train_ds = spark.createDataFrame([            (6.0, Vectors.dense([1.0, 1.0])),            (8.0, Vectors.dense([1.0, 2.0])),            (9.0, Vectors.dense([2.0, 2.0])),            (11.0, Vectors.dense([2.0, 3.0]))        ],            ["label", "features"]        )        # Create model and train.        service = structure_streaming_service.StructuredStreamingService()        service.train(train_ds)        # Predict and results.        def extract_features(x):            values = x.split(",")            features_ = []            for i in values[1:]:                features_.append(float(i))            features = Vectors.dense(features_)            return features        extract_features_udf = udf(extract_features, VectorUDT())        def extract_label(x):            values = x.split(",")            label = float(values[0])            return label        extract_label_udf = udf(extract_label, FloatType())        predict_ds = spark.readStream.format("text").option("path", "data/structured_streaming").load() \            .withColumn("features", extract_features_udf(col("value"))) \            .withColumn("label", extract_label_udf(col("value")))        service.predict(predict_ds).awaitTermination(3)        # +-----+------------------+        # |label|        prediction|        # +-----+------------------+        # |  1.0|15.966990887541273|        # |  2.0|18.961384020443553|        # +-----+------------------+    def setUp(self):        warnings.filterwarnings("ignore", category=ResourceWarning)        warnings.filterwarnings("ignore", category=DeprecationWarning)if __name__ == "__main__":    unittest.main()

Нужно отметить, что для Scala можно воспользоваться созданием потока в памяти.
Это может выглядеть вот так:

implicit val sqlCtx = spark.sqlContextimport spark.implicits._val source = MemoryStream[Record]source.addData(Record(1.0, Vectors.dense(3.0, 5.0)))source.addData(Record(2.0, Vectors.dense(4.0, 6.0)))val predictDs = source.toDF()service.predict(predictDs).awaitTermination(2000)

Полный пример на Scala (здесь, для разнообразия, не используется sql):

package aaa.abc.dd.spark_streaming_pr.clusterimport org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}import org.apache.spark.sql.DataFrameimport org.apache.spark.sql.functions.udfimport org.apache.spark.sql.streaming.StreamingQueryclass StructuredStreamingService {  var service: LinearRegression = _  var model: LinearRegressionModel = _  def train(ds: DataFrame): Unit = {    service = new LinearRegression().setMaxIter(10).setRegParam(0.01)    model = service.fit(ds)  }  def predict(ds: DataFrame): StreamingQuery = {    val m = ds.sparkSession.sparkContext.broadcast(model)    def transformFun(features: org.apache.spark.ml.linalg.Vector): Double = {      m.value.predict(features)    }    val transform: org.apache.spark.ml.linalg.Vector => Double = transformFun    val toUpperUdf = udf(transform)    val predictionDs = ds.withColumn("prediction", toUpperUdf(ds("features")))    predictionDs      .writeStream      .foreachBatch((r: DataFrame, i: Long) => {        r.show()        // scalastyle:off println        println(s"$i")        // scalastyle:on println      })      .start()  }}

Тест:

package aaa.abc.dd.spark_streaming_pr.clusterimport org.apache.spark.ml.linalg.Vectorsimport org.apache.spark.sql.SparkSessionimport org.apache.spark.sql.execution.streaming.MemoryStreamimport org.scalatest.{Matchers, Outcome, fixture}class StructuredStreamingServiceSuite extends fixture.FunSuite with Matchers {  test("run") { spark =>    // Prepare data.    val trainDs = spark.createDataFrame(Seq(      (6.0, Vectors.dense(1.0, 1.0)),      (8.0, Vectors.dense(1.0, 2.0)),      (9.0, Vectors.dense(2.0, 2.0)),      (11.0, Vectors.dense(2.0, 3.0))    )).toDF("label", "features")    // Create model and train.    val service = new StructuredStreamingService()    service.train(trainDs)    // Predict and results.    implicit val sqlCtx = spark.sqlContext    import spark.implicits._    val source = MemoryStream[Record]    source.addData(Record(1.0, Vectors.dense(3.0, 5.0)))    source.addData(Record(2.0, Vectors.dense(4.0, 6.0)))    val predictDs = source.toDF()    service.predict(predictDs).awaitTermination(2000)    // +-----+---------+------------------+    // |label| features|        prediction|    // +-----+---------+------------------+    // |  1.0|[3.0,5.0]|15.966990887541273|    // |  2.0|[4.0,6.0]|18.961384020443553|    // +-----+---------+------------------+  }  override protected def withFixture(test: OneArgTest): Outcome = {    val spark = SparkSession.builder().master("local[2]").getOrCreate()    try withFixture(test.toNoArgTest(spark))    finally spark.stop()  }  override type FixtureParam = SparkSession  case class Record(label: Double, features: org.apache.spark.ml.linalg.Vector)}

Выводы

При написании тестов необходимо разделять код таким образом, чтобы разделять логику и применение конкретных вызовов API. Можно использоваться любые доступные источники. В том числе и kafka.

Такие абстракции как DataFrame позволяют это сделать легко и просто.

При использовании Python данные придется хранить в файлах.

Ссылки и ресурсы

• Исходный код Python
• Исходный код Scala

Открываем предзаказ продвинутого курса по Apache Kafka.

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

Курс подойдет опытным инженерам.

Программа

Базовая часть (будет доступна бесплатно в личном кабинете Слёрма)

Введение

• Немного истории: почему лог и стриминг важны, и занимают значительное место в бизнес процессах компаний;
• Fun fact: почему "кафка"? А не почему, просто хорошо звучит.

Базовые основы технологии

• Сравнения с подобными технологиями;
• Базовые примитивы: брокеры, топики, партиции, оффсеты, retention.

Установка, запись и чтение создание и конфигурация топиков

• Первая запись и первое чтение;
• Базовые основы Apache Zookeeper.

Продвинутая часть

Тема 1: Работа с распределенным кластером

• Конфигурация кластера,
• Отказоустойчивость,
• Контроллер,
• Репликация между кластерами в разных ЦОД,
• Примеры архитектуры.

Тема 2: Клиентские библиотеки

• Producer,
• Consumer,
• Как (не) потерять данные: консистентность против доступности.

Тема 3: Мониторинг

• Ключевые метрики и алерты,
• SLI & SLO.

Тема 4: Анализ производительности

• Почему Кафка такая быстрая?
• Бенчмаркинг.

Тема 5: Поддержка работоспособности кластера и траблшутинг

• Балансировка нагрузки: partition assignment & repartitioning,
• Обновление версии кластера и клиентов,
• Чтение логов,
• Продвинутый функционал траблшутинга,
• Примеры сбоев из жизни.

Тема 6: Развертывание кластера в проде

• Рекомендуемая конфигурация и архитектура,
• Практики и примеры из жизни.

Авторы курса:

• Анатолий Солдатов, Lead Engineer в Avito;
• Александр Миронов, Infrastructure Engineer в Stripe.

Релиз курса в апреле 2021 года. Сейчас собираем обратную связь по программе, готовим материалы и уже записываем бесплатную базовую часть курса. Оставить заявку на весь курс или бесплатные вводные уроки можно здесь.

Kafka повсюду. Где есть микросервисы и распределенные вычисления, а они сейчас популярны, там почти наверняка есть и Kafka. В статье я попытаюсь объяснить, как в Kafka работает механизм хранения.

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

Обычно считается, что Kafka это распределенная и реплицированная очередь сообщений. С технической точки зрения все верно, но термин очередь сообщений не все понимают одинаково. Я предпочитаю другое определение: распределенный и реплицированный журнал коммитов. Эта формулировка кажется более точной, ведь мы все прекрасно знаем, как журналы записываются на диск. Просто в этом случае на диск попадают сообщения, отправленные в Kafka.

Применительно к хранению в Kafka используется два термина: партиции и топики. Партиции это единицы хранения сообщений, а топики что-то вроде контейнеров, в которых эти партиции находятся.

С основной теорией мы определились, давайте перейдем к практике.

Я создам в Kafka топик с тремя партициями. Если хотите повторять за мной, вот как выглядит команда для локальной настройки Kafka в Windows.

kafka-topics.bat --create --topic freblogg --partitions 3 --replication-factor 1 --zookeeper localhost:2181

В каталоге журналов Kafka создано три каталога:

$ tree freblogg*freblogg-0|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpointfreblogg-1|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpointfreblogg-2|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpoint

Мы создали в топике три партиции, и у каждой свой каталог в файловой системе. Еще тут есть несколько файлов (index, log и т д.), но о них чуть позже.

Обратите внимание, что в Kafka топик это логическое объединение, а партиция фактическая единица хранения. То, что физически хранится на диске. Как устроены партиции?

Партиции

В теории партиция это неизменяемая коллекция (или последовательность) сообщений. Мы можем добавлять сообщения в партицию, но не можем удалять. И под мы я подразумеваю продюсеров в Kafka. Продюсер не может удалять сообщения из топика.

Сейчас мы отправим в топик пару сообщений, но сначала обратите внимание на размер файлов в папках партиций.

$ ls -lh freblogg-0total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpoint

Как видите, файлы index вместе весят 20 МБ, а файл log совершенно пустой. В папках freblogg-1 и freblogg-2 то же самое.
Давайте отправим сообщения через console producer и посмотрим, что будет:

kafka-console-producer.bat --topic freblogg --broker-list localhost:9092

Я отправил два сообщения сначала ввел стандартное Hello World, а потом нажал на Enter, и это второе сообщение. Еще раз посмотрим на размеры файлов:

$ ls -lh freblogg*freblogg-0:total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpointfreblogg-1:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  68 Aug  5 10:15 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 10:15 leader-epoch-checkpointfreblogg-2:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  79 Aug  5 09:59 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 09:59 leader-epoch-checkpoint

Два сообщения заняли две партиции, и файлы log в них теперь имеют размер. Это потому, что сообщения в партиции хранятся в файле xxxx.log. Давайте заглянем в файл log и убедимся, что сообщение и правда там.

$ cat freblogg-2/*.log@^@^B^@^K^X^@^@^@^A"^@^@^A^VHello World^@

Файлы с форматом log не очень удобно читать, но мы все же видим в конце Hello World, то есть файл обновился, когда мы отправили сообщение в топик. Второе сообщение мы отправили в другую партицию.

Обратите внимание, что первое сообщение попало в третью партицию (freblogg-2), а второе во вторую (freblogg-1). Для первого сообщения Kafka выбирает партицию произвольно, а следующие просто распределяет по кругу (round-robin). Если мы отправим третье сообщение, Kafka запишет его во freblogg-0 и дальше будет придерживаться этого порядка. Мы можем и сами выбирать партицию, указав ключ. Kafka хранит все сообщения с одним ключом в одной и той же партиции.

Каждому новому сообщению в партиции присваивается Id на 1 больше предыдущего. Этот Id еще называют смещением (offset). У первого сообщения смещение 0, у второго 1 и т. д., каждое следующее всегда на 1 больше предыдущего.

<Небольшое отступление>

Давайте используем инструмент Kafka, чтобы понять, что это за странные символы в файле log. Нам они кажутся бессмысленными, но для Kafka это метаданные каждого сообщения в очереди. Выполним команду:

kafka-run-class.bat kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files logs\freblogg-2\00000000000000000000.log

Получим результат:

umping logs\freblogg-2\00000000000000000000.logStarting offset: 0offset: 0 position: 0 CreateTime: 1533443377944 isvalid: true keysize: -1 valuesize: 11 producerId: -1 headerKeys: [] payload: Hello Worldoffset: 1 position: 79 CreateTime: 1533462689974 isvalid: true keysize: -1 valuesize: 6 producerId: -1 headerKeys: [] payload: amazon

(Я удалил из выходных данных кое-что лишнее.)

Здесь мы видим смещение, время создания, размер ключа и значения, а еще само сообщение (payload).

</Небольшое отступление>

Надо понимать, что партиция привязана к брокеру. Если у нас, допустим, три брокера, а папка freblogg-0 существует в broker-1, в других брокерах ее не будет. У одного топика могут быть партиции в нескольких брокерах, но одна партиция всегда существует в одном брокере Kafka (если установлен коэффициент репликации по умолчанию 1, но об этом чуть позже).

Сегменты

Что это за файлы index и log в каталоге партиции? Партиция, может, и единица хранения в Kafka, но не минимальная. Каждая партиция разделена на сегменты, то есть коллекции сообщений. Kafka не хранит все сообщения партиции в одном файле (как в файле лога), а разделяет их на сегменты. Это дает несколько преимуществ. (Разделяй и властвуй, как говорится.)

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

$ ls -lh freblogg-0total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpoint

Нули (00000000000000000000) в файлах log и index в каждой папке партиции это имя сегмента. У файла сегмента есть файлы segment.log, segment.index и segment.timeindex.

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

В имени каждого файла сегмента отражается смещение от первого сообщения. На картинке выше в сегменте 0 содержатся сообщения со смещением от 0 до 2, в сегменте 3 от 3 до 5, и так далее. Последний сегмент, шестой, сейчас активен.

$ ls -lh freblogg*freblogg-0:total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpointfreblogg-1:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  68 Aug  5 10:15 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 10:15 leader-epoch-checkpointfreblogg-2:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  79 Aug  5 09:59 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 09:59 leader-epoch-checkpoint

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

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

Допустим, мы отправили в партицию freblogg-2 три сообщения, и она выглядит так:

Три сообщения это наш лимит. На следующем сообщении Kafka автоматически закроет текущий сегмент, создаст новый, сделает его активным и сохранит новое сообщение в файле log этого сегмента. (Я не показываю предыдущие нули, чтобы было проще воспринять).

freblogg-2|-- 00.index|-- 00.log|-- 00.timeindex|-- 03.index|-- 03.log|-- 03.timeindex`--

Удивительное дело, но новый сегмент называется не 01. Мы видим 03.index, 03.log. Почему так?

Kafka называет сегмент по имени минимального смещения в нем. Новое сообщение в партиции имеет смещение 3, поэтому Kafka так и называет новый сегмент. Раз у нас есть сегменты 00 и 03, мы можем быть уверены, что сообщения со смещениями 0, 1 и 2 и правда находятся в сегменте 00. Новые сообщения в партиции freblogg-2 со смещениями 3 ,4 и 5 будут храниться в сегменте 03.

В Kafka мы часто читаем сообщения по определенному смещению. Искать смещение в файле log затратно, особенно если файл разрастается до неприличных размеров (по умолчанию это 1 ГБ). Для этого нам и нужен файл .index. В файле index хранятся смещения и физическое расположение сообщения в файле log.

Файл index для файла log, который я приводил в кратком отступлении, будет выглядеть как-то так:

Если нужно прочитать сообщение со смещением 1, мы ищем его в файле index и видим, что его положение 79. Переходим к положению 79 в файле log и читаем. Это довольно эффективный способ мы быстро находим нужное смещение в уже отсортированном файле index с помощью бинарного поиска.

Параллелизм в партициях

Чтобы гарантировать порядок чтения сообщений из партиции, Kafka дает доступ к партиции только одному консюмеру (из группы консюмеров). Если партиция получает сообщения a, f и k, консюмер читает их в том же порядке: a, f и k. Это важно, ведь порядок потребления сообщений на уровне топика не гарантирован, если у вас несколько партиций.

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

Топики

Наконец, переходим к топикам. Мы уже кое-что знаем о них. Главное, что нужно знать: топик это просто логическое объединение нескольких партиций.

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

Репликация

Как работает репликация? Создавая топик в Kafka, мы указываем для него коэффициент репликации replication-factor. Допустим, у нас два брокера и мы устанавливаем replication-factor 2. Теперь Kafka попытается всегда создавать бэкап, или реплику, для каждой партиции в топике. Kafka распределяет партиции примерно так же, как HDFS распределяет блоки данных по нодам.

Допустим, для топика freblogg мы установили коэффициент репликации 2. Мы получим примерно такое распределение партиций:

Даже если реплицированная партиция находится в другом брокере, Kafka не разрешает ее читать, потому что в каждом наборе партиций есть LEADER, то есть лидер, и FOLLOWERS ведомые, которые остаются в резерве. Ведомые периодически синхронизируются с лидером и ждут своего звездного часа. Когда лидер выйдет из строя, один из in-sync ведомых будет выбран новым лидером, и вы будете получать данные из этой партиции.

Лидер и ведомый одной партиции всегда находятся в разных брокерах. Думаю, не нужно объяснять, почему.

Мы подошли к концу этой длинной статьи. Если вы добрались до этого места поздравляю. Теперь вы знаете почти все о хранении данных в Kafka. Давайте повторим, чтобы ничего не забыть.

Итоги

• В Kafka данные хранятся в топиках.
• Топики разделены на партиции.
• Каждая партиция разделена на сегменты.
• У каждого сегмента есть файл log, где хранится само сообщение, и файл index, где хранится позиция сообщения в файле log.
• У одного топика могут быть партиции в разных брокерах, но сама партиция всегда привязана к одному брокеру.
• Реплицированные партиции существуют пассивно. Вы обращаетесь к ним, только если сломался лидер.

От редакции:
Более подробно от работе с Apache Kafka можно узнать на курсе Слёрма. Курс сейчас в разработке. В программе бесплатные базовые уроки и платная продвинутая часть. Оставьте заявку на курс, чтобы получать уведомления.

Ресурсы:

Схема Kafka https://kafka.apache.org/images/kafka_diagram.png

Когда сервисы интегрируются при помощи Kafka очень удобно использовать REST API, как универсальный и стандартный способ обмена сообщениями. При увеличении количества сервисов сложность коммуникаций увеличивается. Для контроля можно и нужно использовать интеграционное тестирование. Такие библиотеки как testcontainers или EmbeddedServer прекрасно помогают организовать такое тестирование. Существуют много примеров для micronaut, Spring Boot и т.д. Но в этих примерах опущены некоторые детали, которые не позволяют с первого раза запустить код. В статье приводятся примеры с подробным описанием и ссылками на код.

Пример

Для простоты можно принять такой REST API.

/runs POST-метод. Инициализирует запрос в канал связи. Принимает данные и возвращает ключ запроса.
/runs/{key}/status GET-метод. По ключу возвращает статус запроса. Может принимать следующие значения: UNKNOWN, RUNNING, DONE.
/runs /{key} GET-метод. По ключу возвращает результат запроса.

Подобный API реализован у livy, хотя и для других задач.

Реализация

Будут использоваться: micronaut, Spring Boot.

micronaut

Контроллер для API.

import io.micronaut.http.annotation.Body;import io.micronaut.http.annotation.Controller;import io.micronaut.http.annotation.Get;import io.micronaut.http.annotation.Post;import io.reactivex.Maybe;import io.reactivex.schedulers.Schedulers;import javax.inject.Inject;import java.util.UUID;@Controller("/runs")public class RunController {    @Inject    RunClient runClient;    @Inject    RunCache runCache;    @Post    public String runs(@Body String body) {        String key = UUID.randomUUID().toString();        runCache.statuses.put(key, RunStatus.RUNNING);        runCache.responses.put(key, "");        runClient.sendRun(key, new Run(key, RunType.REQUEST, "", body));        return key;    }    @Get("/{key}/status")    public Maybe<RunStatus> getRunStatus(String key) {        return Maybe.just(key)                .subscribeOn(Schedulers.io())                .map(it -> runCache.statuses.getOrDefault(it, RunStatus.UNKNOWN));    }    @Get("/{key}")    public Maybe<String> getRunResponse(String key) {        return Maybe.just(key)                .subscribeOn(Schedulers.io())                .map(it -> runCache.responses.getOrDefault(it, ""));    }}

Отправка сообщений в kafka.

import io.micronaut.configuration.kafka.annotation.*;import io.micronaut.messaging.annotation.Body;@KafkaClientpublic interface RunClient {    @Topic("runs")    void sendRun(@KafkaKey String key, @Body Run run);}

Получение сообщений из kafka.

import io.micronaut.configuration.kafka.annotation.*;import io.micronaut.messaging.annotation.Body;import javax.inject.Inject;@KafkaListener(offsetReset = OffsetReset.EARLIEST)public class RunListener {    @Inject    RunCalculator runCalculator;    @Topic("runs")    public void receive(@KafkaKey String key, @Body Run run) {        runCalculator.run(key, run);    }}

Обработка сообщений происходит в RunCalculator. Для тестов используется особая реализация, в которой происходит переброска сообщений.

import io.micronaut.context.annotation.Replaces;import javax.inject.Inject;import javax.inject.Singleton;import java.util.UUID;@Replaces(RunCalculatorImpl.class)@Singletonpublic class RunCalculatorWithWork implements RunCalculator {    @Inject    RunClient runClient;    @Inject    RunCache runCache;    @Override    public void run(String key, Run run) {        if (RunType.REQUEST.equals(run.getType())) {            String runKey = run.getKey();            String newKey = UUID.randomUUID().toString();            String runBody = run.getBody();            runClient.sendRun(newKey, new Run(newKey, RunType.RESPONSE, runKey, runBody + "_calculated"));        } else if (RunType.RESPONSE.equals(run.getType())) {            runCache.statuses.replace(run.getResponseKey(), RunStatus.DONE);            runCache.responses.replace(run.getResponseKey(), run.getBody());        }    }}

Тест.

import io.micronaut.http.HttpRequest;import io.micronaut.http.client.HttpClient;import static org.junit.jupiter.api.Assertions.assertEquals;public abstract class RunBase {    void run(HttpClient client) {        String key = client.toBlocking().retrieve(HttpRequest.POST("/runs", "body"));        RunStatus runStatus = RunStatus.UNKNOWN;        while (runStatus != RunStatus.DONE) {            runStatus = client.toBlocking().retrieve(HttpRequest.GET("/runs/" + key + "/status"), RunStatus.class);            try {                Thread.sleep(500);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        String response = client.toBlocking().retrieve(HttpRequest.GET("/runs/" + key), String.class);        assertEquals("body_calculated", response);    }}

Для использования EmbeddedServer необходимо.

Подключить библиотеки:

testImplementation("org.apache.kafka:kafka-clients:2.6.0:test")testImplementation("org.apache.kafka:kafka_2.12:2.6.0")testImplementation("org.apache.kafka:kafka_2.12:2.6.0:test")

Тест может выглядеть так.

import io.micronaut.context.ApplicationContext;import io.micronaut.http.client.HttpClient;import io.micronaut.runtime.server.EmbeddedServer;import org.junit.jupiter.api.Test;import java.util.HashMap;import java.util.Map;public class RunKeTest extends RunBase {    @Test    void test() {        Map<String, Object> properties = new HashMap<>();        properties.put("kafka.bootstrap.servers", "localhost:9092");        properties.put("kafka.embedded.enabled", "true");        try (EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer.class, properties)) {            ApplicationContext applicationContext = embeddedServer.getApplicationContext();            HttpClient client = applicationContext.createBean(HttpClient.class, embeddedServer.getURI());            run(client);        }    }}

Для использования testcontainers необходимо.

Подключить библиотеки:

implementation("org.testcontainers:kafka:1.14.3")

Тест может выглядеть так.

import io.micronaut.context.ApplicationContext;import io.micronaut.http.client.HttpClient;import io.micronaut.runtime.server.EmbeddedServer;import org.junit.jupiter.api.Test;import org.testcontainers.containers.KafkaContainer;import org.testcontainers.utility.DockerImageName;import java.util.HashMap;import java.util.Map;public class RunTcTest extends RunBase {    @Test    public void test() {        try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.3"))) {            kafka.start();            Map<String, Object> properties = new HashMap<>();            properties.put("kafka.bootstrap.servers", kafka.getBootstrapServers());            try (EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer.class, properties)) {                ApplicationContext applicationContext = embeddedServer.getApplicationContext();                HttpClient client = applicationContext.createBean(HttpClient.class, embeddedServer.getURI());                run(client);            }        }    }}

Spring Boot

Контроллер для API.

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.*;import java.util.UUID;@RestController@RequestMapping("/runs")public class RunController {    @Autowired    private RunClient runClient;    @Autowired    private RunCache runCache;    @PostMapping()    public String runs(@RequestBody String body) {        String key = UUID.randomUUID().toString();        runCache.statuses.put(key, RunStatus.RUNNING);        runCache.responses.put(key, "");        runClient.sendRun(key, new Run(key, RunType.REQUEST, "", body));        return key;    }    @GetMapping("/{key}/status")    public RunStatus getRunStatus(@PathVariable String key) {        return runCache.statuses.getOrDefault(key, RunStatus.UNKNOWN);    }    @GetMapping("/{key}")    public String getRunResponse(@PathVariable String key) {        return runCache.responses.getOrDefault(key, "");    }}

Отправка сообщений в kafka.

import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.stereotype.Component;@Componentpublic class RunClient {    @Autowired    private KafkaTemplate<String, String> kafkaTemplate;    @Autowired    private ObjectMapper objectMapper;    public void sendRun(String key, Run run) {        String data = "";        try {            data = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(run);        } catch (JsonProcessingException e) {            e.printStackTrace();        }        kafkaTemplate.send("runs", key, data);    }}

Получение сообщений из kafka.

import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;@Componentpublic class RunListener {    @Autowired    private ObjectMapper objectMapper;    @Autowired    private RunCalculator runCalculator;    @KafkaListener(topics = "runs", groupId = "m-group")    public void receive(ConsumerRecord<?, ?> consumerRecord) {        String key = consumerRecord.key().toString();        Run run = null;        try {            run = objectMapper.readValue(consumerRecord.value().toString(), Run.class);        } catch (JsonProcessingException e) {            e.printStackTrace();        }        runCalculator.run(key, run);    }}

Обработка сообщений происходит в RunCalculator. Для тестов используется особая реализация, в которой происходит переброска сообщений.

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.UUID;@Componentpublic class RunCalculatorWithWork implements RunCalculator {    @Autowired    RunClient runClient;    @Autowired    RunCache runCache;    @Override    public void run(String key, Run run) {        if (RunType.REQUEST.equals(run.getType())) {            String runKey = run.getKey();            String newKey = UUID.randomUUID().toString();            String runBody = run.getBody();            runClient.sendRun(newKey, new Run(newKey, RunType.RESPONSE, runKey, runBody + "_calculated"));        } else if (RunType.RESPONSE.equals(run.getType())) {            runCache.statuses.replace(run.getResponseKey(), RunStatus.DONE);            runCache.responses.replace(run.getResponseKey(), run.getBody());        }    }}

Тест.

import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.http.MediaType;import org.springframework.test.web.servlet.MockMvc;import org.springframework.test.web.servlet.MvcResult;import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;import static org.junit.jupiter.api.Assertions.assertEquals;import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;public abstract class RunBase {    void run(MockMvc mockMvc, ObjectMapper objectMapper) throws Exception {        MvcResult keyResult = mockMvc.perform(MockMvcRequestBuilders.post("/runs")                .content("body")                .contentType(MediaType.APPLICATION_JSON)                .accept(MediaType.APPLICATION_JSON))                .andExpect(status().isOk())                .andReturn();        String key = keyResult.getResponse().getContentAsString();        RunStatus runStatus = RunStatus.UNKNOWN;        while (runStatus != RunStatus.DONE) {            MvcResult statusResult = mockMvc.perform(MockMvcRequestBuilders.get("/runs/" + key + "/status")                    .contentType(MediaType.APPLICATION_JSON)                    .accept(MediaType.APPLICATION_JSON))                    .andExpect(status().isOk())                    .andReturn();            runStatus = objectMapper.readValue(statusResult.getResponse().getContentAsString(), RunStatus.class);            try {                Thread.sleep(500);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        String response = mockMvc.perform(MockMvcRequestBuilders.get("/runs/" + key)                .contentType(MediaType.APPLICATION_JSON)                .accept(MediaType.APPLICATION_JSON))                .andExpect(status().isOk())                .andReturn().getResponse().getContentAsString();        assertEquals("body_calculated", response);    }}

Для использования EmbeddedServer необходимо.

Подключить библиотеки:

<dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka</artifactId>    <version>2.5.10.RELEASE</version></dependency><dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka-test</artifactId>    <version>2.5.10.RELEASE</version>    <scope>test</scope></dependency>

Тест может выглядеть так.

import com.fasterxml.jackson.databind.ObjectMapper;import org.junit.jupiter.api.Test;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.boot.test.context.TestConfiguration;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Import;import org.springframework.kafka.test.context.EmbeddedKafka;import org.springframework.test.web.servlet.MockMvc;@AutoConfigureMockMvc@SpringBootTest@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})@Import(RunKeTest.RunKeTestConfiguration.class)public class RunKeTest extends RunBase {    @Autowired    private MockMvc mockMvc;    @Autowired    private ObjectMapper objectMapper;    @Test    void test() throws Exception {        run(mockMvc, objectMapper);    }    @TestConfiguration    static class RunKeTestConfiguration {        @Autowired        private RunCache runCache;        @Autowired        private RunClient runClient;        @Bean        public RunCalculator runCalculator() {            RunCalculatorWithWork runCalculatorWithWork = new RunCalculatorWithWork();            runCalculatorWithWork.runCache = runCache;            runCalculatorWithWork.runClient = runClient;            return runCalculatorWithWork;        }    }}

Для использования testcontainers необходимо.

Подключить библиотеки:

<dependency>    <groupId>org.testcontainers</groupId>    <artifactId>kafka</artifactId>    <version>1.14.3</version>    <scope>test</scope></dependency>

Тест может выглядеть так.

import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.apache.kafka.common.serialization.StringSerializer;import org.junit.ClassRule;import org.junit.jupiter.api.Test;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.boot.test.context.TestConfiguration;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Import;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.core.*;import org.springframework.test.web.servlet.MockMvc;import org.testcontainers.containers.KafkaContainer;import org.testcontainers.utility.DockerImageName;import java.util.HashMap;import java.util.Map;@AutoConfigureMockMvc@SpringBootTest@Import(RunTcTest.RunTcTestConfiguration.class)public class RunTcTest extends RunBase {    @ClassRule    public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.3"));    static {        kafka.start();    }    @Autowired    private MockMvc mockMvc;    @Autowired    private ObjectMapper objectMapper;    @Test    void test() throws Exception {        run(mockMvc, objectMapper);    }    @TestConfiguration    static class RunTcTestConfiguration {        @Autowired        private RunCache runCache;        @Autowired        private RunClient runClient;        @Bean        ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();            factory.setConsumerFactory(consumerFactory());            return factory;        }        @Bean        public ConsumerFactory<Integer, String> consumerFactory() {            return new DefaultKafkaConsumerFactory<>(consumerConfigs());        }        @Bean        public Map<String, Object> consumerConfigs() {            Map<String, Object> props = new HashMap<>();            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");            props.put(ConsumerConfig.GROUP_ID_CONFIG, "m-group");            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);            return props;        }        @Bean        public ProducerFactory<String, String> producerFactory() {            Map<String, Object> configProps = new HashMap<>();            configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);            return new DefaultKafkaProducerFactory<>(configProps);        }        @Bean        public KafkaTemplate<String, String> kafkaTemplate() {            return new KafkaTemplate<>(producerFactory());        }        @Bean        public RunCalculator runCalculator() {            RunCalculatorWithWork runCalculatorWithWork = new RunCalculatorWithWork();            runCalculatorWithWork.runCache = runCache;            runCalculatorWithWork.runClient = runClient;            return runCalculatorWithWork;        }    }}

Перед всеми тестами необходимо стартовать kafka. Это делается вот таким вот образом:

kafka.start();

Дополнительные свойства для kafka в тестах можно задать в ресурсном файле.

application.yml

spring:  kafka:    consumer:      auto-offset-reset: earliest

Ресурсы и ссылки

Код для micronaut

Код для Spring Boot

PART 1: TESTING KAFKA MICROSERVICES WITH MICRONAUT

Testing Kafka and Spring Boot

Micronaut Kafka

Spring for Apache Kafka

Мотивом для написания данной статьи послужил тот факт, что на habr.com участилось появление материалов маркетингового характера про Apache Kafka. А также тот факт, что из статей складывается впечатление что пишут их немного далекие от реального использования люди это конечно же только впечателение, но почему-то в большинстве своем статьи обязательно содержат сравнение Apache Kafka с RabbitMQ, причем не в пользу последнего. Что самое интересное читая подобные статьи управленцы без технического бэкграунда начинают тратить деньги на внутренние исследования, чтобы ведущие разработчики и технические директора выбрали одно из решений. Так как я очень жадный/домовитый, а также так как я сторонник тезиса "В споре НЕ рождается истина" предлагаю вам ознакомится с другим подходом почти без сравнения разных брокеров.

Без сравнения никуда

Вообще, по правильному, я должен был сделать статью в формате Kafka+RabbitMQ+Nats+ActiveMQ+Mosquito+etc, но мне кажется, что для Вас дорогие читатели это будет перебор, хотя обычно в моих архитектурных решениях присутствуют все вышеуказанные сервисы (и не только). И это я еще не рассказываю про AzureServiceBus/AmazonServiceBus которые также участвуют в "гибридах" при крупных программах проектов. Поэтому пока остановимся на связке Kafka+RabbitMQ и далее вы поймете почему: по аналогии можно подключить любой сервис с его протоколом. Потому что:

сравнивая Apache Kafka и RabbitMQ вы сравниваете 2 (два) бренда, а точнее 2 коммерческие компании Confluent и vmWare, и немножко Apache Software Foundation (но это не компания)

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

• RabbitMQ мультипротокольный и расширяемый брокер сообщений
• Apache Kafka платформа для распределенной потоковой передачи событий
• Confluent Platform платформа потоковой передачи событий с возможностью создания высокопроизводительных конвейеров обработки данных для целей аналитики и интеграции в бизнес-сценариях

Я не зря третьим пунктом выделяю наработки компании Confluent те кто собирается использовать Apache Kafka в продуктиве должны хотя бы видеть какую функциональность дополнительно добавляет Confluent к Apache Kafka. А это SchemeRegistry, RestProxy, kSQL и еще несколько интересных штук, о одной из которых мы поговорим ниже, она называется Kafka-Connect.

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

Итак чтобы было совсем понятно, куда я веду.

• ключевая особенность RabbitMQ мультипротокольность и расширяемость. (основной язык якобы Erlang)
• ключевая особенность экосистемы Kafka потоковая передача с обработкой (основной язык якобы Scala/Java)

Отсюда и возникают минусы каждого из решений

• для RabbitMQ мы не сможем построить нормального решения для потоковой обработки. Точнее сможем, но НЕ штатно.
• а для Kafka мы не сможем сделать мультипротокольность, точнее сможем но НЕ штатно.

Сократ не говорил, что в споре рождается истина

Еще одна новость: действительно если почитать источник, то Сократ вообще-то в итоге пришел к тому, что нужно обеспечить диалог, а если по научному то истина рождается в научном споре, который формально представляет собой процесс публикация со ссылкой на источники -> а затем научная критика оппонентов -> истина

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

• ODBC
• AMQP
• MSMQ
• XMPP
• IP over Avian Carriers

так как тогда наша задача была интегрировать всякое (python, C#, java) и 1С был придуман проект One-S-Connectors (https://code.google.com/archive/p/one-c-connectors/source/default/source). Сейчас он имеет сугубо академический интерес (так как в 1С мире моя персона достаточно известна и на Хабре много 1С специалистов из сообщества "воинствующих 1С-ников" эта ссылка специально для них).
Однако уже тогда (в 2006 году) стало понятно, что по большому счету конечному разработчику придется менять/выбирать протокол под бизнес-задачу. А инфраструктурщикам придется обеспечить максимально широкий спектр интеграционных протоколов. От ODBC до Kafka/NATs/ModBus.

Но вернемся к дню сегодняшнему когда я начал использовать в проектах уровня ГИС (госсударственные информационные системы) различные транспорта данных внезапно выяснилось, что универсальные адаптеры это не только концепт воинствующих 1С-ников, но и соседей. Поэтому многие идеи при внедрении черпались из еще двух интересных проектов

• библиотека Kombu (для Python) https://docs.celeryproject.org/projects/kombu/en/stable/introduction.html#transport-comparison
• комплект CAP для .NetCore https://github.com/dotnetcore/CAP

маленькое примечание для менеджеров про Kombu как то так получилось, что имплементация протокола Apache Kafka до сих пор открыта https://github.com/celery/kombu/issues/301 и почему-то перешла в разряд "Дайте денег", поэтому для Python проектов приходится использовать дополнительно https://github.com/confluentinc/confluent-kafka-python

Когда вы дочитаете до этого момента предполагаю, что вы зададите вопрос про остальные языки: Java, GoLang, RUST, etc. Но во первых я не зря выше указал что по серьезному в наш обсуждаемый сегодня гибрид нужно добавить историю про NATs и ActiveMQ и внезапно JMS поэтому просьба дочитать до конца: Java будет, а во вторых мы переходим к еще трем полезным ссылкам

• https://github.com/rabbitmq/rabbitmq-server/tree/master/deps/
• https://docs.confluent.io/current/connect/kafka-connect-rabbitmq/index.html
• https://github.com/84codes/kafka-connect-rabbitmq/blob/master/docker-compose.yml

Прокоментируем их? Дело в том, что как бы вы не хотели, а для полноценного использования "в длинную" вам придется подписаться на историю релизов как сервера RabbitMQ и самое главное на те самые расширения (лежат в каталоге /deps) которые постоянно добавляются в ядро RabbitMQ, так и на портал компании Confluent где она публикует приложения полезные для конечного бизнеса использующего Apache Kafka в продуктиве.

подход к расширяемости за счет активируемых расширений также используется в экосистеме PostgreSQL тот который CREATE EXTENSION hypopg, так что подход реализованный компанией Pivotal/vmWare далеко не новый в нашем чудесном мире архитектуры программного обеспечения

Дополнительно на чудесном рынке облачных услуг в формате "Серьезная штука как сервис" есть еще один игрок это компания 84Codes https://github.com/84codes. Когда в рамках проектов внедрения нет нормальных инженеров по инфраструктуре именно 84Codes спасает пилотные проекты, потому как у них можно легко арендовать бесплатные/сильнодешевые контура CloudAMQP и CloudKarafka.

Я как бы обещал, что не буду ничего говорить про деньги, однако придется отразить 2 ключевых момента:

• компания vmWare зарабатывает известно на чем, поэтому RabbitMQ ей развивается как часть своей платформы то есть они инвестируют в открытый проект не особо занимаясь его монетизацией. Возврат их инвестиций происходит в других местах, ну и также за счет контрибьторов на GitHub.
• а вот компания Confuent собирается монетизировать свою платформу через Enterprise лицензию в которую включает те самые коннекторы Enterprise-Kafka-Connect, а также GUI для управления платформой.

Когда-то давно существовал https://github.com/jcustenborder/kafka-connect-rabbitmq, примечателен тот факт что товарищ Джереми его скрыл, оставив только свои наработки для Java разработчиков в виде Maven Archetype https://github.com/jcustenborder/kafka-connect-archtype еще раз обращаю Ваше внимание, что компания Confluent будет и дальше пытаться монетизировать свою деятельность, так что переводить всю интеграцию только на Kafka я бы на вашем месте поостерегся.

Поэтому когда вам топят за Kafka учитывайте, что вы либо изучаете Java, либо платите за Enterprise лицензию. А когда вам топят за RabbitMQ учитывайте, что либо вы изучаете системное администрирование (Erlang накладывает особенности системного администрирования), либо покупаете сервис у провайдеров типа 84Codes. Кодить на Erlang никогда не придется там это не нужно, если только вы не контрибьюторы OpenStack.

Поставил и забыл уже не работает

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

• использование только одного протокола интеграции приводит к появлению ProtocolLock и как следствие к VendorLock я же не зря выше написал, что за каждым открытым продуктом, стоит какой-то ключевой комплект вендоров как они себя поведут: мы не знаем.
• в мире ИТ больше нет серьезных продуктов, которые бы представляли собой монолитную службу все приложения давно стали композитными.
• все нормальные вендоры сокращают свои релизные циклы по ключевым продуктам нормальной практикой стало выпускать редакции раз в 3 месяца TDD, BDD, CICD, ScallableAgile и DevOps (DocOps, DevSecOps) эти инженерные практики и методики управления не просто так развиваются. Всем очень хочется сокращать себестоимость и TimeToMarket.

Абзац выше важен, как финальный аккорд, прежде чем мы перейдем к Docker-Compose. А именно к нему я вел чтобы и разработчики и инфраструктурщики понимали что такое гибридная инфраструктура в режиме мультипротокольности (с) нужно сделать так, чтобы каждый мог поэкспериментировать с предлагаемым контуром. Как я уже указал выше первично подобное применительно к Kafka+RabbitMQ было подсмотрено именно у коллег из 84Codes (хорошие ребята всем советую https://www.84codes.com/).

Чтобы вы смогли поэкспериментировать сами

Итак подходим к примерам, так как обоснования и вводных уже хватит. Предположим вы уже поняли, что вам также нужна мультипротокольность, однако мы же помним, что все рекламные материалы про Apache Kafka нам рассказывают что это единственное решение с реализацией exactly-ones доставки сообщений от отправителя получателю. Собственно на самом деле нам и нужен гибрид, чтобы сделать из связки ТочкаОбмена->Очередь журнал Kafka (это тот который Topic) чтобы возникла сущность под называнием Offsets у нашей очереди событий.

Предлагаю попробовать. Концепт для проверки Вашими руками будет состоять из:

• Zookeper
• KafkaBroker
• RabbitMQ
• KafkaConnect

и трех приложений приложений

• отправитель на Python по протоколу AMQP 0.9
• получатель на С# по протоколу AMQP 1.0
• получатель на C# по протоколу Kafka

Еще интересное замечание: когда вы смотрите на всякие обучающие видео по Apache Kafka авторы часто (но не всегда) старательно пишут примеры на Java, это они делают скорее всего для того, чтобы скрыть от вас особенности использования librdkafka C++ библиотеки на основе которой сделаны многие не-джава адаптеры,. Я же наоборот предлагаю вам начинать исследование интеграции с Kafka именно с неё, чтобы четко оценивать риски "куда вы ввязываетесь": очень примечательно что там работает фактически один разработчик, формально в одиночку https://github.com/edenhill/librdkafka/pulse/monthly, а допустим wmWare старается поддерживать свою линейку клиентов под своим брендом https://github.com/rabbitmq

ну и самое главное и тяжелое:

контур содержит открытый форк старого RabbitMQ-Kafka-Sinc-Connector того самого который товарищи из Confluent в своё время скрыли с Github.

Докер контура для экспериментов

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

Развертываем RabbitMQ и Kafka

контур инфраструктуры который нам понадобится запускается достаточно просто

docker-compose -f dockers/infra.yml up -d

Если вам интересно что же там внутри, нашего композитного приложения, то в конце статьи дается ссылка на полный комплект исходников, наиболее интересен в нем Kafka-UI и непосредственно RabbitMQ-Sinc, все остальное обычно и штатно для всех известных примеров по Kafka или RabbitMQ

    image: provectuslabs/kafka-ui:latest    ports:      - 8080:8080    depends_on:      - kafka-broker      - zookeeper    environment:      KAFKA_CLUSTERS_0_NAME: local      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: broker:29092      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181      KAFKA_CLUSTERS_0_JMXPORT: 9101

Но самое главное кроется в репозитории Java

    <parent>        <groupId>com.github.jcustenborder.kafka.connect</groupId>        <artifactId>kafka-connect-parent</artifactId>        <version>1.0.0</version>    </parent>

Если подробно изучить pom.xml то выяснится, что существует заглавный проект для всех конекторов к Кафка https://github.com/jcustenborder/kafka-connect-parent, в котором используется Java-Kafka-Adapter

И непосредственно синхронизацией c RMQ занимается штатный Java клиент https://www.rabbitmq.com/java-client.html

            <groupId>com.rabbitmq</groupId>            <artifactId>amqp-client</artifactId>            <version>${rabbitmq.version}</version>

Таким образом по правильному, чтобы получились повторить тот же эксперимент что и у меня, необходимо выполнить:

• собрать из исходников java синхронизатор -1-build-connect-jar.bat
• собрать контейнер с синхрозатором 00-build-connect-image.sh

и уже потом запустить полный инфраструктурный контур

• стартуем полный инфраструктурный контур 01-start-infra.sh

обратите внимание так как Docker использует разное поведение при работе с PWD для Windows и Linux приходится делать дубликаты скриптов. В остальных случаях под обоими операционными системами используется интерпретатор sh

В итоге вы получите следующий комплект сервисов

На картинке можно увидеть как подключаются конфигурационные файлы к RabbitMQ и какая топология сетевых портов у нас будет участвовать в эксперименте:

Назначение портов:

• 9092 будет использоваться для Kafka протокола
• 8080 используется для отображения красивой картинки состояния Apache Kafka UI
• 5672 будет использоваться для протокола AMQP 0.9 и он же будет работать и как AMQP 1.0
• 15672 используется для красивой картинки управления RabbitMQ
• 28082 отладочный порт для управления через curl трансформатором протоколов

В этот момент нужно остановится и прокомментировать особенность развертывания RabbitMQ в Docker:

• хорошей практикой является версионирование включенных плагинов расширений enabled-rmq-plugins

[    rabbitmq_management,     rabbitmq_amqp1_0,     rabbitmq_mqtt,     rabbitmq_federation,     rabbitmq_federation_management,    rabbitmq_shovel,    rabbitmq_shovel_management,    rabbitmq_prometheus].

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

     "bindings":[        {           "source":"orders-send",           "vhost":"/",           "destination":"orders-amqp-10-consumer",           "destination_type":"queue",           "routing_key":"",           "arguments":{

Запускаем наши приложения

Остается только запустить наши приложения эмулирующие подключения

docker-compose -f dockers/infra.yml restart protocol-connect-syncdocker-compose -f applications.yml builddocker-compose -f applications.yml up

Топология наших тестовых приложений достаточно простая

Исходный код также максимально упрощён:

• отправляется как-будто бы заказ Васи с периодичностью в 2 секунды

        producer = conn.Producer(serializer='json')        producer.publish({'client': 'Вася', 'count': 10, 'good': 'АйФончик'},                      exchange=order_exchange,                      declare=[kafka_queue, amqp10_queue])        time.sleep(2)

RUN python -m pip install \    kombu \    librabbitmq

причем используется для этого максимально производительная библиотека на Си для AMQP 0.9 librabbitmq наследуется именно от неё https://github.com/alanxz/rabbitmq-c

• создан подписчик который уже по протоколу AMQP 1.0 смотрит в свою очередь и получает события, соответственно очередь очищается и больше мы заказов Васи не получим. В этом потоке нам это и не нужно.

            Attach recvAttach = new Attach()            {                Source = new Source()                {                    Address = "orders-amqp-10-consumer",                    Durable = 1,                },

            ReceiverLink receiver =                 new ReceiverLink(session,"netcore_amqp_10_consumer", recvAttach, null);            Console.WriteLine("Receiver connected to broker.");            while (true) {                Message message = receiver.Receive();                if (message == null)                {                    Console.WriteLine("Client exiting.");                    break;                }                Console.WriteLine("Received "                   + System.Text.Encoding.UTF8.GetString((byte[])message.Body)

Причем в качестве драйвера выбран

  <ItemGroup>    <PackageReference Include="AMQPNetLite.Core" Version="2.4.1" />  </ItemGroup>

именно его https://github.com/Azure/amqpnetlite Microsoft использует для маркетинга своей реализации сервисной шины. Собственно именно AMQP 1.0 как протокол они и рекламируют https://docs.microsoft.com/ru-ru/azure/service-bus-messaging/service-bus-amqp-overview

Ну и финально

• создан подписчик по протоколу Kafka который при каждом старте перечитывает с нуля журнал отправленных заказов Васи. Тот самый Exactly once.

                AutoOffsetReset = AutoOffsetReset.Earliest

                c.Subscribe("orders-from-amqp");

                    while (true)                    {                        try                        {                            var cr = c.Consume(cts.Token);

Выглядит наш контур в итоге следующим образом:

• 5 инфраструктурных контейнеров

• 3 контейнера с приложениями

• готовый журнал транзакций заказов который можно посмотреть через Kafka-Ui

• и готовый контур связей для RabbitMQ

А где же Java ?

Не волнуйтесь при таком гибридном подходе, без неё никуда, для того чтобы всё вышеуказанное заработало пришлось сделать форк и актуализировать версии Kafka-Connect-Base

[submodule "dockers/rabbitmq-kafka-sink"]    path = dockers/rabbitmq-kafka-sink    url = https://github.com/aliczin/kafka-connect-rabbitmq

Но самое интересное не это, самое интересное что в этом самом Kafka-Connect нет по сути никакой магии только код трансформации.

По сути нам предлагают:

• создать наследника абстрактной задачи Источника

public class RabbitMQSourceTask extends SourceTask {

• выполнить подписку на очередь сообщений

        this.channel.basicConsume(queue, this.consumer);        log.info("Setting channel.basicQos({}, {});", this.config.prefetchCount, this.config.prefetchGlobal);        this.channel.basicQos(this.config.prefetchCount, this.config.prefetchGlobal);

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

  @Override  public List<SourceRecord> poll() throws InterruptedException {    List<SourceRecord> batch = new ArrayList<>(4096);    while (!this.records.drain(batch)) {

Отдельно можно выделить чудесный трансформатор сообщений из AMQP 0.9 в Кафка. У несведующего в Java глаз может задергаться. У автора чувствуется многолетный опыт работы в J2EE.

  private static final Logger log = LoggerFactory.getLogger(MessageConverter.class);  static final String FIELD_ENVELOPE_DELIVERYTAG = "deliveryTag";  static final String FIELD_ENVELOPE_ISREDELIVER = "isRedeliver";  static final String FIELD_ENVELOPE_EXCHANGE = "exchange";  static final String FIELD_ENVELOPE_ROUTINGKEY = "routingKey";  static final Schema SCHEMA_ENVELOPE = SchemaBuilder.struct()      .name("com.github.jcustenborder.kafka.connect.rabbitmq.Envelope")      .doc("Encapsulates a group of parameters used for AMQP's Basic methods. See " +          "`Envelope <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Envelope.html>`_")      .field(FIELD_ENVELOPE_DELIVERYTAG, SchemaBuilder.int64().doc("The delivery tag included in this parameter envelope. See `Envelope.getDeliveryTag() <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Envelope.html#getDeliveryTag-->`_").build())      .field(FIELD_ENVELOPE_ISREDELIVER, SchemaBuilder.bool().doc("The redelivery flag included in this parameter envelope. See `Envelope.isRedeliver() <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Envelope.html#isRedeliver-->`_").build())      .field(FIELD_ENVELOPE_EXCHANGE, SchemaBuilder.string().optional().doc("The name of the exchange included in this parameter envelope. See `Envelope.getExchange() <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Envelope.html#getExchange-->`_"))      .field(FIELD_ENVELOPE_ROUTINGKEY, SchemaBuilder.string().optional().doc("The routing key included in this parameter envelope. See `Envelope.getRoutingKey() <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Envelope.html#getRoutingKey-->`_").build())      .build();

Но Не будем критиковать, мы же в самом начале договорились что наша главная задача добиться конечного результата удобным на сегодня способом. А итоги у нас получаются следующие.

Итоги

Все что здесь продемонстрировано естественно лежит на Github.

В репозитории https://github.com/aliczin/hybrid-eventing. Лицензия выставленна простая до невозможности Creative Commons Attribution 4.0 International.

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

Схема коммуникации в итоге для "разработчика интеграционных потоков" (с) выглядит следующим образом для источника и брокеров

orderEventsApp->Amqp09: send orderAmqp09->Amqp10: fanout\n copy eventAmqp09->KafkaQ: fanout\n copy eventKafkaQ->KafkaConnect: consume\n on messageKafkaConnect->KafkaConnect: transform\n messageKafkaConnect->Kafka: publish to topic

а для приемников все упрощается

Amqp10->orderEventSubApp: subcribe\n for eventorderJournalApp->Kafka: read kafka journal

Приемники берут нужные им данные только по нужному им протоколу

Ключевые посылы

Ключевые моменты которые я хотел расскрыть данной статьей

• стройте эксперименты и продуктивы с Apache Kafka не со штатным Java клиентом, а librdkafka и базирующихся на ней адаптерах это позволит вам отладить сценарии разных версий протоколов KafkaAPI. Java вам пригодится в другом месте.

  
  

• не ввязывайтесь с священные войны, что лучше RabbitMQ/Kafka/Nats/ActiveMQ просто развертывайте сервисы и публикуйте протоколы и пробуйте свои бизнес-сценарии.

  
  

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

  
  

• реальный ИТ ландшафт почти всегда будет мультипротокольным

  
  

Примечание для понимающих

чтобы гибриды развивались дальше:

• Mosquito очень удобен как встраиваемый брокер на уровне контролера SCADA для преобразования из ModBus/OPC-UA. Хотя как вы уже поняли из статьи интересны реализации "мостов из протокола в протокол" пример https://github.com/mainflux/mainflux

  
  

• ActiveMQ удобен для Java разработчиков, потому что у них есть боязнь Erlang, но как мы выше уже сказали мост RabbitMQ AMQP 1.0 -> ActiveMQ легко организуется средствами RabbitMQ, кстати также как и JMS.

  
  

• NATs интересен как часть OpenFaaS платформы, при внедрении "своего маленького" Amazon Lambda с преферансом. И опять же подход будет всё тот же мультипротокольные мосты с трансформацией: https://github.com/nats-io/nats-kafka если Вам не страшно посмотрите эксперименты с OpenFaaS веселых 1С-ников 2.5 часа примеров https://youtu.be/8sF-oGGVa9M

  
  

Надеюсь мой архитектурный подход Вам придется по душе и вы перестанете тратить деньги заказчика (инвестора/свои если вы стартапщик: Маша это замечание специально для тебя) на бессмысленные обсуждения что же выбрать в качестве брокера/платформы, и начнете наконец-то делать функциональность, которая будет использовать тот протокол, который удобен прямо сейчас. С возможностью переключения в случае "если чё"

Функциональность: Мультипротокольный адаптер    Как разработчик я хочу иметь абстракцию Produser/Consumer    С возможность изменения протокола интеграции    Чтобы под каждую задачу выбирать разные протоколы     и единый интерфейс вызова для обеспечения независимости от вендора предоставляющего транспортСценарий: vmWare реализует протокол Stream средствами RabbitMQ     Когда vmWare закончит свой плагин для потоков    Тогда я активирую новый протокол     И быстро воткну его в приложение    И так как у меня есть продуктивный кластер RabbitMQ    И мне нужно будет просто поменять канал для отдельных бизнес сценариевСценарий: Завтра придут 1С-ники со своим ActiveMQ из Шины для 1С    Когда мне нужно быстро включить очереди 1С в общий контур    И чтобы на Питоне использовать старые наработки с Kafka API    Тогда я добавляю трансформацию ActivemeMQ2Kafka    и живу по старому а события ходят уже и из 1Сetc

А чтобы вы не думали, что данный подход это нечто уникальное вот Вам еще интересная ссылка: https://github.com/fclairamb/ftpserver/pull/34 это когда нужен FTP сервер, а хочется S3.

Ну и в качестве финального момента обратите внимание: есть и риски данного подхода: но они я думаю Вам и так понятны.

• Придется оркестрировать такой комплект сервисов и вручную это почти невозможно. Придется использовать DevOps штуки типа k8s, OpenShift, etc но если вы уже решились на интеграцию в режимах слабой связаности приложений в режиме онлайн, у вас что-то на эту тему уже скорее всего есть.
• Трансформаторы между протоколами приходится дорабатывать ничего готового открытого и PRODUCTION-READY на данный момент найти почти невозможно.

Финальное примечение для любителей писать ТЗ по ГОСТу

так как Хабр читают любители цифровой трансформации (чтобы кто не понимал под этим словом) советую в техническое задание добавлять не упоминание конкретных реализации серверов, а что-то примерно следующее:

комплект программ для интеграции должен реализовывать коммуникацию конечных приложений по открытым протоколам HTTP, AMQP 0.9, AMQP 1.0, Apache Kafka не ниже версии 23, MQTT, WebSockets, <ЛюбойДругойХотьSOAPХотяЭтоЖуть> с возможность преобразования между протоколами дополнительными средствами администрирования

Надеюсь моя публикация после долгого перерыва Вам будет полезна в ваших интеграционных проектах. Предполагаю что будет вопрос про 1С и тут у меня совет только один. Используйте Google по ключевым словам 1С+RabbitMQ или 1С+Kafka или 1С+OpenFaas и RabbitMQ и Kafka "в 1С" давно и непринужденно используются. Потому что 1С это не только язык, но и несколько сообществ где уже давно сделаны все возможные адаптеры и платные и бесплатные. Собственно как и в Java/C#/Python/C++/Rust/etc.

Данная статья написана с применением расширения https://shd101wyy.github.io/markdown-preview-enhanced для Visual Studio Code за что автору летят дополнительные лучи добра.

Ну и в качестве финального момента хотел бы заметить, что выбор Cunfluent Inc в качестве платформы разработки Kafka-Connect экосистемы JDK выглядит все таки странно. Не удивлюсь если их конкуренты сделают такое же, но на GoLang, NodeJS (что-нибудь типа Kafka-Beats-Hub)

Красивые картинки в формате GraphViz я делаю при помощи хитрого проекта Docker2GraphViz помогает поддерживать актуальный контур и техническую документацию в формате Markdown

set CURPATH=%~dp0set DOCKER_DIR=%CURPATH%\dockersdocker run --rm -it --name dcv -v %DOCKER_DIR%\:/input pmsipilot/docker-compose-viz render -m image --force --output-file=infra-topology.png infra.ymldocker run --rm -it --name dcv -v %CURPATH%\:/input pmsipilot/docker-compose-viz render -m image --force --output-file=apps-topology.png applications.ymlcopy /b/v/y dockers\infra-topology.png content\assets\infra-topology.pngcopy /b/v/y apps-topology.png content\assets\apps-topology.png