Версия сайта для слабовидящих
Недавно мы писали об изменении статуса и улучшении протокола KRaft в Apache Kafka 3.3. Сегодня погрузимся в эту тему чуть глубже и рассмотрим, как отказ от Zookeeper влияет на количество разделов и возможность одного и того же кластера Kafka с одним набором топиков обслуживать разные типы приложений в различных бизнес-сценариях.
Еще раз про Zookeeper и количество разделов в Apache Kafka
Начиная с выпуска 3.3 режим KRaft, который позволяет обойтись без внешнего сервиса синхронизации метаданных Zookeeper, что мы разбирали здесь, приобрел статус готовности для новых кластеров Apache Kafka. Это означает, что его можно использовать в производственных развертываниях. В релизе 3.3.2 были ввдены некоторые улучшения этого режима: ограничено время ожидания heartbeat-сигнала брокера KRaft не более значения конфигурации broker.session.timeout.ms. Брокеры KRaft поддерживают свою работоспособность в кластере, отправляя запросы BROKER_HEARTBEAT на активный контроллер, который изолирует брокера при отсутствии такого сигнала в течение периода, определенного в конфигурации broker.session.timeout.ms. Брокер должен использовать тайм-аут запроса для своих запросов BROKER_HEARTBEAT, который не превышает тайм-аут сеанса, используемый контроллером. Это сделано, чтобы предотвратить недостаточную репликацию разделов из-за сбоя активного контроллера и задержки heartbeat-сигналов брокеров на новый контроллер.
Таким образом, между количеством разделов и работой KRaft, а ранее Zookeeper, есть закономерная связь. В частности, именно они устанавливают ограничения на количество разделов, разрешенных в кластере. Например, максимальное количество разделов на брокера в AWS MSK — 4000. Разделы и их ключи определяют параллелизм, возможный в топике Kafka. В каждой группе потребителей Kafka может быть только один потребитель, который читает из определенного раздела. Таким образом, когда в кластере Kafka потребительская нагрузка распределена неравномерно, это может быть связано с неправильным ключом раздела, специфичным для приложения. Репликация разделов настраивается в кластере. Каждый раздел имеет настроенное количество реплик, а также ведущую реплику и резервные реплики. Запись и чтение происходят через ведущую реплику, поэтому рекомендуется распределять реплики ведущего раздела между брокерами, чтобы свести к минимуму нагрузку перераспределения реплик в случае выхода из строя одного из брокеров. Подробнее про группы потребителей Kafka читайте в нашей прошлой статье.
Клиенты Kafka, т.е. приложения-продюсеры, потребители и AdminClient, используют узлы контроллера Kafka и Zookeeper для отслеживания метаданных. Это увеличивает накладные расходы на синхронизацию и управление клиентами Kafka, узлами Kafka Controller и Zookeeper. Ранее все метаданные (смещение фиксации, назначение группового раздела и пр), хранились в Zookeeper, который поддерживал карту всех брокеров в кластере Kafka и их статус. Когда брокер присоединялся к кластеру или покидал его, Zookeeper отслеживал эти изменения и передавать их другим узлам в кластере. Обновления узла Zookeeper и контроллера были синхронны, а обновления брокеров — асинхронны, что приводило к так называемому состоянию гонки. Так называют ситуацию конкурирования за ресурсы в распределенных системах, когда несколько потоков одновременно обращаются к одному и тому же ресурсу, причём хотя бы один из них выполняет операцию записи, и порядок этих обращений точно не определен.
При запуске брокер он пытается пометить себя как контроллер, сообщая об этом Zookeeper, который отвечает на это сообщением «Контроллер уже доступен». Когда брокер становится недоступным для Zookeeper, он помечается как недоступный и удаляет запись. Если брокер снова становится доступным для Zookeeper, он должен повторно получить всю информацию о кластере. Таким образом, при выходе из строя или перезапуске узла контроллера, он должен прочитать все метаданные для всех брокеров и разделов из Zookeeper, а затем отправить эти метаданные всем брокерам. Это становится почти невозможным, когда раздел на брокере становится недоступным ни для узла контроллера, ни для Zookeeper.
Чтобы устранить это ограничение, а также снизить зависимость от внешнего по отношению к Kafka сервису, еще в 2019 году разработчики этой распределенной платформы потоковой передачи событий начали работу по замене Zookeeper. В 2021 году вышла альтернатива – протокол консенсуса KRaft (Kafka Raft). Что он собой представляет и как решает проблему количества разделов, рассмотрим далее.
Что такое KRaft и как он работает
Raft — это алгоритм консенсуса, эквивалентный Paxos (семейство протоколов для решения задачи консенсуса в сети ненадёжных вычислителей) по отказоустойчивости и производительности. Консенсус является фундаментальной проблемой отказоустойчивых распределенных систем и предполагает согласование значений несколькими серверами. Как только они примут решение о значении, это решение станет окончательным. Типичные алгоритмы консенсуса достигают успеха, когда доступно любое большинство их серверов. Например, кластер из 5 серверов может продолжать работу даже при выходе из строя 2 серверов. Если больше серверов выйдет из строя, они перестанут работать, но никогда не вернут неверный результат, обеспечивая согласованность данных, но жертвуя доступностью.
Консенсус обычно возникает в контексте реплицированных конечных автоматов — общего подхода к построению отказоустойчивых систем. Каждый сервер имеет конечный автомат и журнал. Конечный автомат — это компонент системы, который надо сделать отказоустойчивым, например, хеш-таблица. Клиенты взаимодействуют с конечным автоматом так, будто он очень надежен, даже если меньшинство серверов в кластере выйдет из строя. Каждый конечный автомат принимает на вход команды из своего журнала. В примере с хеш-таблицей журнал будет включать такие команды, как установить x равным y. Алгоритм консенсуса используется для согласования команд в журналах серверов и должен гарантировать, что если какой-либо конечный автомат применит set x to y в качестве n-й команды, ни один другой конечный автомат никогда не применит другую n-ю команду. В результате каждый конечный автомат обрабатывает одну и ту же серию команд и, выдает одну и ту же серию результатов, и так приходит к одной и той же серии состояний.
Raft — это алгоритм консенсуса, который разбивается на относительно независимые подзадачи и четко решает все основные части для прикладных систем. Поэтому в рамках легендарного KIP-500 разработчики Kafka представили новый протокол на основе консенсуса под названием KRaft, в котором метаданные хранятся в виде журнала фиксации. Вместо хранения метаданных о назначении разделов, смещении фиксации и пр. в Zookeeper, начиная с релиза 2.8 Kafka использует внутренние топики, такие как __offsets. А внутренние API, такие как OffsetCommit, OffsetFetch, JoinGroup, SyncGroup, Heartbeat и т. д., теперь обрабатываются самой Kafka, а не отправляются в Zookeeper. Брокеры могут получать сообщения в журналах и обрабатывать их, постоянно обновляясь по мере обработки сообщений. Если брокер становится недоступным в течение определенного времени, после его запуска он может обрабатывать сообщения из внутренних топиков, а после обработки сообщений он снова может быть помечен как активный. Необходимо обрабатывать только дельта-сообщения, т.е. появившиеся за период отсутствия брокера в кластере. Меньший объем данных значительно сокращает время для повторного присоединения к кластеру, ускоряя работу всей Big Data системы.
Определенные узлы контроллера поддерживают самоуправляемый кворум для определения лидера. Если ведущий контроллер выйдет из строя, резервный контроллер может почти мгновенно взять на себя управление, как только его сообщения будут обновлены. Все это снижает нагрузку на партиционирование и поддержание метаданных в больших кластерах Apache Kafka.
Возвращаясь к особенностям хранения данных в этой распределенной платформе потоковой передачи событий, отметим, что все сообщения в Kafka хранятся в лог-файлах на жестком диске. Размер потребляемого пространства и скорость работы дисков зависят от периода хранения, настроенного для данных в конфигурациях log.retention.ms и/или retention.bytes, о чем мы писали здесь. Благодаря базовой способности Kafka воспроизводить сообщения с самого начала, различные приложения используют возможность настройки более длительных периодов хранения.
Но длительные периоды хранения увеличивают нагрузку на диск с точки зрения резервного копирования и миграции данных, повышая риски для потребителей в том же кластере, которые зависят от относительно недавних сообщений и не нуждаются в долгом хранении сообщений. Для решения этой проблемы разработчики Kafka предложили концепцию локального и удаленного уровней хранения.
Период хранения и тип диска можно настроить отдельно для разных уровней. Например, для уровня локального хранилища можно использовать меньший SSD-диск емкостью 100 ГБ со сроком хранения 2 дня. А для удаленного уровня выбрать HBase или S3 со сроком хранения 6 месяцев. Для приложений, которым нужны более старые данные, чем данные в локальном хранилище, брокеры Kafka сами будут обращаться к удаленному уровню и получать данные оттуда. Это позволяет одному и тому же кластеру Kafka с одним набором топиков обслуживать разные типы приложений.
Освойте администрирование и эксплуатацию Apache Kafka и Flink для потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
- Администрирование кластера Kafka
- Apache Kafka для инженеров данных
- Администрирование Arenadata Streaming Kafka
Источники
