23 января 2023 года вышел очередной релиз самой популярной платформы потоковой передачи событий. Разбираемся с новинками Apache Kafka 3.3.2: готовность протокола KRaft, новый API для метрик, разделитель по умолчанию для записей без ключа, исправления и улучшения, важные для дата-инженера и администратора кластера.
Apache Kafka 3.3.2: главные новинки и изменения
Минорный выпуск 3.3.2 носит технический характер: в нем всего 3 улучшения и чуть более 20 исправленных ошибок. Перед тем, как перейти к этим изменениям, перечислим наиболее заметные изменения в линейке 3.3. Прежде всего, это готовность к эксплуатации для новых кластеров режима KRaft, который позволяет обойтись без внешнего сервиса синхронизации метаданных Zookeeper, о чем мы писали здесь. Поэтому процесс обновления для кластеров Apache Kafka с KRaft немного отличается от обновления для кластеров на основе Zookeeper.
Также улучшен разделитель, используемый по умолчанию для записей без ключей, чтобы избежать патологического поведения, когда один или несколько брокеров работают медленно. Новая логика может повлиять на поведение пакетной обработки, которое можно настроить с помощью параметров конфигурации batch.size и linger.ms. Предыдущее поведение можно восстановить, установив параметру partitioner.class значение org.apache.kafka.clients.producer.internals.DefaultPartitioner.
Еще из заметных новинок стоит отметить новый API addMetricIfAbsent для системных метрик. Он создает новую метрику, если она не существует, или возвращает существующую, если она уже зарегистрирована. Это поведение отличается от API-интерфейса addMetric, который генерирует исключение IllegalArgumentException при попытке создать уже существующую метрику.
Теперь рассмотрим 3 улучшения выпуска 3.3.2. Одним из них является получение ответа об ошибке при обращении к публичному провайдеру аутентификации OAuth/OIDC. Класс org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever используется для отправки учетных данных клиента публичному провайдеру OAuth/OIDC и получения ответа, включая токен доступа. Теперь при ошибке подробные сведения извлекаются со стороны провайдера и помогают клиенту определить, что сбой при получении токена вызван неправильной конфигурацией на его стороне.
Другим важным улучшением стало ограничение времени ожидания heartbeat-сигнала брокера KRaft не более значения конфигурации broker.session.timeout.ms. Брокеры KRaft поддерживают свою работоспособность в кластере, отправляя запросы BROKER_HEARTBEAT на активный контроллер, который изолирует брокера при отсутствии такого сигнала в течение периода, определенного в конфигурации broker.session.timeout.ms. Брокер должен использовать тайм-аут запроса для своих запросов BROKER_HEARTBEAT, который не превышает тайм-аут сеанса, используемый контроллером. До версии Apache Kafka 3.3.2 при отработке отказа контроллера брокер мог не вовремя отменить существующий запрос контрольного сигнала, а затем передать его новому контроллеру для поддержания непрерывного сеанса в кластере. Таким образом, сбой активного контроллера приводил к тому, что разделы будут недостаточно реплицированы или ниже минимального количества согласованных реплик (ISR) просто из-за задержки heartbeat-сигналов брокеров на новый контроллер. Теперь это исправлено. Подробнее про особенности работы протокола KRaft читайте в нашей новой статье.
Наконец, согласованы порты для чтения системных метрик. Ранее, при наличии брэндмауэра или работы внутри контейнера, раскрытие только com.sun.management.jmxremote.port не позволяло получить метрики, а без указания порта RMI он генерировался случайным образом по умолчанию. В Apache Kafka 3.3.2 оба порта стали согласованными для чтения указанных метрик.
Из исправленных ошибок наиболее важными можно назвать следующие:
- обновление тайм-аута перебалансировки при повторном присоединении статического потребителя в группе;
- задержка отключения, управляемого брокером KRaft, на неопределенный срок;
- бесконечный цикл psend без ключа записи и batch.size=0;
- невозможность удалить топики, в названии которых есть точка (.);
- проблемы с выбором реплики для чтения при обновлении метаданных при перебалансировке;
- некорректный кодировщик Base64 в OAuthBearerLoginCallbackHandler при OIDC-аутентификации;
- обработка неудачной выборки при не назначенных разделах.
Как перейти на новый релиз с предыдущих версий: стратегия и тактика обновления
При обновление до 3.3.x с любой версии от 0.8.x до 3.2.x, важно помнить про несовместимые изменения и невозможность отката назад из-за схемы, используемой для хранения смещений потребителей и параметра inter.broker.protocol.version. Для непрерывного обновления следует обновить server.properties для всех брокеров и добавить следующие свойства:
- CURRENT_KAFKA_VERSION – текущая версия, с которой выполняется обновление.
- CURRENT_MESSAGE_FORMAT_VERSION – текущая версия формата сообщения. Если она была переопределена ранее, следует сохранить ее текущее значение. При обновлении с версии до 0.11.0.x, CURRENT_MESSAGE_FORMAT_VERSION должен быть установлен в соответствии с CURRENT_KAFKA_VERSION.
- broker.protocol.version=CURRENT_KAFKA_VERSION, например, 3.2, 3.1 и пр.;
- message.format.version=CURRENT_MESSAGE_FORMAT_VERSION
При обновлении с версии 0.11.0.x или выше без переопределения формата сообщения, нужно переопределить только версию межброкерского протокола.
При обновлении отдельно взятого брокера кластера Apache Kafka, его следует выключить, обновить код кинфигураций и перезапустить. После этого брокер будет работать с последней версией, и можно убедиться, что поведение и производительность кластера соответствуют ожиданиям. На этом этапе все еще можно понизить версию в случае возникновения каких-либо проблем. После проверки поведения и производительности кластера надо изменить версию протокола, отредактировав inter.broker.protocol.version и установив для нее значение 3.3. После поочередного перезапуска брокеров одного за другим новая версия протокола вступит в силу. Как только все брокеры кластера начнут использовать последнюю версию протокола, понизить ее до более старой версии станет невозможно.
При переопределении версии формата сообщения, необходимо выполнить еще один последовательный перезапуск, чтобы обновить его. Когда все или большинство потребителей будут обновлены до версии 0.11.0 или более поздней, следует изменить log.message.format.version на 3.3 на каждом брокере и перезапустить их один за другим. Старые клиенты Scala, которые больше не поддерживаются, не поддерживают формат сообщений, представленный в версии 0.11. Поэтому, чтобы избежать затрат на преобразование или воспользоваться семантикой строго однократной доставки, необходимо использовать более новые клиенты Java.
При обновлении кластера на базе KRaft до версии 3.3.x с любой версии от 3.0.x до 3.2.x после изменения metadata.version на последнюю версию, перейти на версию до 3.3-IV0 будет невозможно. Для непрерывного обновления следует последовательно выключить каждый брокер, обновить код конфигурации и перезапустить его. После проверки поведения и производительности кластера Apache Kafka надо изменить версию метаданных, запустив команду./bin/kafka-features.sh upgrade —metadata 3.3. Версию метаданных кластера нельзя понизить до предварительной версии 3.0.x, 3.1.x или 3.2.x после ее обновления, но можно перейти на рабочие версии, такие как 3.3-IV0, 3.3-IV1 и т. д.
Освойте администрирование и эксплуатацию Apache Kafka и Flink для потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
- Администрирование кластера Kafka
- Apache Kafka для инженеров данных
- Администрирование Arenadata Streaming Kafka
Источники
