Offset explorer как пользоваться

Documentation

If your cluster is configured for plaintext security (typically in test environments only) you do not need to configure any additional security attributes. You can just click on Test to test that your connection is working properly or Add to add the server connection without testing it first.

If your cluster is configured for SASL (plaintext or SSL) you must either specify the JAAS config in the UI or pass in your JAAS config file to Offset Explorer when you start it. The exact contents of the JAAS file depend on the configuration of your cluster, please refer to the Kafka documentation.

offsetexplorer.exe -J-Djava.security.auth.login.config=c:/client_jaas.conf

offsetexplorer -J-Djava.security.auth.login.config=/client_jaas.conf

/Applications/Offset Explorer.app/Contents/java/app

If you do not pass the JAAS config file at the startup of Offset Explorer, you must specify it under the JAAS Config tab for each connection.

If you are using SASL Plaintext you typically must change the sasl.mechanism client property to PLAIN. This property can be entered in the ‘SASL Mechanism’ text field under the ‘Advanced’ section.

If your Kafka cluster is configured to use SSL you may need to set various SSL configuration parameters. Unless your Kafka brokers are using a server certificate issued by a public CA, you need to point to a local truststore that contains the self signed root certificate that signed your brokers certificate. You also need to enter the password for this truststore.

If the SAN(s) in your server certificate do not match the actual hostname of the brokers you are connecting to, you will receive an SSL error (No subject alternative DNS name matching xxx found) when you try to connect. You can avoid this by unchecking the ‘Validate SSL endpoint hostname’ checkbox in the ‘Broker security’ section. This will set the ssl.endpoint.identification.algorithm client property to null.

If your Kafka cluster requires a client certificate (two-way authentication) you also need to configure your keystore attributes. The keystore contains the private key that you use to authenticate to your Kafka brokers. You also need to configure a password for the keystore as well as password for the private key in the keystore.

Apache Kafka. Безопасность, мониторинг и управление кластером

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

Почему Apache Kafka?

Apache Kafka — это унификация. Десятки поставщиков и потребителей, миллионы сообщений в день и огромные массивы данных — для этого нужна надежная, отказоустойчивая и высокопроизводительная шина данных. Существует множество версий дистрибутивов Apache Kafka, например vanilla kafka, oбразы от confluent, bitnami, wurstmeister и т.д. Мы расскажем про решение на базе сборок от Confluent в виде Docker-образов. Оно самое надежное в плане информационной безопасности. Конфигурирование и запуск контейнеров с Kafka в данном случае происходит с помощью docker-compose.

За время работы с Kafka мы прошли путь от «коробочного» решения до тонкой настройки безопасности кластера, применения TLS шифрования и разворачивания по рекомендациям вендора по построению геораспределенного кластера. В конфигурации «из коробки» нет первоначальных настроек безопасности. Для контроля доступов, вносимых изменений и безопасного взаимодействия перед нами встала задача настройки на кластере Apache Kafka TLS шифрования, аутентификации и авторизации средствами встроенного ACL.

TLS шифрование

Обеспечивает шифрование соединения между брокерами Kafka, серверами Zookeeper, клиентами и брокерами. Вся ключевая информация помещается в хранилища — keystore, которые разделяют на два вида:

keystore, где хранятся ключи и сертификаты стороны, в отношении которой пройдет процедура аутентификации и установления защищенного соединения, например, приватный ключ и ассоциированный с ним и подписанный со стороны центра сертификации (ЦC) сертификат.

truststore, куда помещаются сертификаты, которым доверяет носитель хранилища.

Общий процесс настройки TLS выглядит так:

Для каждого сервера брокера и Zookeeper создаем keystore (настройки конфигурации см. в следующей главе):

Создаем keystore + генерируем ключевую пару — закрытый ключ и соответствующую цепь сертификатов (certificate chain), в которой после выполнения команды будет самоподписанный сертификат, содержащий его открытый ключ.

Создаем запрос на подпись для ЦC.

Подписываем запрос закрытым ключом ЦC. В нашем случае есть промежуточный ЦC и корневой ЦC.

Подписанный запрос импортируем в созданный ранее keystore, в цепочку сертификатов закрытого ключа. Сначала импортируем сертификаты всех ЦС, подписавших наш запрос, и только потом сам подписанный запрос. Это необходимо для точного установления цепи сертификатов, которая производится утилитой keytool, иначе получим ошибку.

Производим настройки конфигурации брокеров, серверов Zookeeper и клиентов.

В итоге на брокере и серверах Zookeeper мы получим keystore с 3 ключевыми объектами:

сертификат промежуточного ЦC, подписанный корневым ЦC;

самоподписанный сертификат корневого ЦC;

приватный ключ сервера со связанной цепью сертификатов: серверный сертификат, подписанный промежуточным ЦC;

сертификат промежуточного ЦC, подписанный корневым ЦC и самоподписанный сертификат корневого ЦC.

Создаем truststore для клиентов и брокеров (для взаимодействия в качестве клиента с хостами Zookeeper) и в него импортируем сертификаты ЦC, чьим подписям клиент должен доверять.

Здесь односторонняя аутентификация, т.е. клиент проверяет брокера и Zookeeper серверы брокера. Если нужно настроить двухстороннюю аутентификацию, проверку клиента или Zookeeper серверы на брокере, проделываем те же операции, что и на брокере, по созданию keystore на клиенте, а на Zookeeper создаем соответствующий truststore.

Важно отметить, что при создании клиентов и при использовании библиотеки librdkafka сертификаты доверенных ЦС могут браться клиентом непосредственно из операционной системы (ОС). К примеру, для машин на Windows необходимо на клиенте задавать настройку:

По умолчанию предполагается только Root и если сертификат сервера подписан еще и промежуточным ЦС, клиент не сможет установить цепь доверия и получит ошибку, поэтому нужно указывать CA.

В дистрибутиве Apache Kafka от Confluent предусмотрена директория для размещения файлов, необходимых для настройки TLS. Внутри контейнера это /etc/kafka/secrets . Для размещения файлов необходимо в конфигурационном файле деплоя образов прописать признак монтирования данной директории.

Далее рассмотрим настройку брокера и сервера Apache Zookeeper в составе кластера Apache Kafka. На остальных представителях кластера конфигурация аналогичная.

Настройка keystore

Создание Keystore на стороне серверов кластера

На сервере должна быть установлена Java, также можно выполнить команды внутри Docker контейнера с дистрибутивом Apache Kafka от Confluent:

keytool -genkeypair -validity days -keyalg RSA -keystore server1.keystore.p12 -storetype pkcs12 -alias server1 -dname «CN=fqdn,OU=x,O=x,L=Moscow,ST=Moscow,C=RU» -ext SAN=DNS:fqdn

Где days – срок действия сертификата в днях, fqdn – доменное имя хоста.

Таким образом, создан keystore с названием «server1.keystore.p12» с типом pkcs12, псевдонимом server1 пары приватный\публичный ключ сервера.

Если установлена версия Java 9 и выше, тип -storetype pkcs12 можно не указывать, т.к. он задан по умолчанию. Далее нужно подписать сертификат сервера в данном keystore.

Создание запроса на подпись сертификата

keytool -certreq -keystore server1.keystore.p12 -alias server1 -file server1.csr

Запрос необходимо подписать на стороне ЦС.

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

keytool -importcert -file root-ca.crt -keystore server1.keystore.p12 -alias RootCA

keytool -importcert -file subroot-ca.crt -keystore server1.keystore.p12 -alias SubCA

keytool -importcert -file server1.pem -keystore server1.keystore.p12 -alias server1

Для проверки успешности импорта выводим содержимое keystore командой:

keytool -list -v -keystore server1.keystore.p12

Keystore готов к работе.

Создание truststore

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

keytool -importcert -file root-ca.crt -keystore client.truststore.p12 -alias RootCA

keytool -importcert -file subroot-ca.crt -keystore client.truststore.p12 -alias SubCA

keytool -list -v -keystore client.truststore.p12

Как уже упоминалось ранее, при построении клиента с помощью библиотеки librdkafka будут использоваться доверенные сертификаты ОС.

Настройки конфигурации сервера

Kafka поддерживает протокол SASL для аутентификации. При этом он может использоваться наряду с шифрованием – SASL_SSL, либо без него – SASL_PLAINTEXT. SASL позволяет применять разные механизмы аутентификации: PLAIN, SCRAM, OAUTHBEARER, Kerberos. С ними можно ознакомиться по ссылке.

Конфигурация брокера

Рассмотрены необходимые корректировки в конфигурацию и параметры для запуска кластера Apache Kafka в виде контейнеризированного решения от Сonfluent с использованием docker-compose. Особенности преобразования конфигурационных параметров Kafka для docker-compose файлов деплоя — на официальном сайте Confluent. Параметры для брокеров предваряются — « KAFKA_ », для Zookeeper — « ZOOKEEPER_ ».

Задаем параметр взаимодействия между брокерами через протокол SASL_SSL:

Параметр взаимодействия с клиентами через протокол SASL_SSL:

Необходимо добавить порт, к примеру, 9096 для приема подключений от клиентов с использованием TLS: SASL_SSL

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

KAFKA_ZOOKEEPER_CONNECT: zookeeper_server1.domain.ru:2281, zookeeper_server2.domain.ru:2281, zookeeper_server3.domain.ru:2281

Задаем относительный путь до файла keystore (как уже упоминалось ранее, для confluent стандартный рабочий каталог для настроек TLS — /etc/kafka/secrets), файла с паролем (server1.creds) от keystore и пароля закрытого ключа, который совпадает с паролем от keystore. И аналогичные настройки для файла truststore.

KAFKA_SSL_TRUSTSTORE_PASSWORD:pass
Активация TLS шифрования для взаимодействия брокеров с кластером Zookeeper:

TLS не поддерживается фрэймворком Zookeeper NIO (используется по умолчанию), поэтому переопределяем его на Netty:

Нужно задать расположение паролей keystore и truststore для взаимодействия с Zookeeper. Примечательно, для данных параметров нужно указать абсолютный путь внутри контейнера, т.к. стандартное расположение «/etc/kafka/secrets» автоматически не учитывается:

Конфигурация Zookeeper сервера

Взаимодействие между серверами кластера Zookeeper предлагается осуществлять с использованием TLS:

Как отметили ранее, для активации TLS необходимо переопределить используемый Zookeeper серверами фрэймворк, а также задать keystore, truststore для взаимодействия кворума:

Поднимаем порт, принимающий подключения с использованием TLS:

Задаем keystore и truststore для взаимодействия с клиентами:

Для настройки TLS использовалась утилита keytool .

При настройке keystore, файлов и директорий, необходимых внутри контейнера, нужно учитывать, что в версиях дистрибутива от 5.5.х Kafka и Zookeeper от Confluent работа внутри контейнера происходит от пользователя “appuser”, а не от “root”, как это было раньше, и необходимо выдавать соответствующие права на файлы на уровне системы хоста. Для хостовой системы пользователь appuser представлен UID 1000 и GID 1000.

В сборке Kafka и Zookeeper от Confluent версии 5.0.x установлена Java 8, начиная с версии 6.0.x и в текущей 7.0.x — Java 11. Соответствие версий дистрибутивов Apache Kafka и Confluent приведены на сайте. Данный аспект обуславливает использование ключей утилиты keytool.

К примеру, в официальной документации для создания keystore используется ключ “ –genkey ”. Как гласит документация, данный ключ устарел и переименован в “ –genkeypair ”, начиная с версии Java 6, а « –import » переименован в « –importcert ». При использовании “ –genkey ” без указания « -storetype pkcs12 » в версиях до Java 9 созданный keystore будет использовать признанный устаревшим формат JKS. Исходя из этого, лучше использовать новые ключи для формирования keystore.

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

В зависимости от политики и правил подписания сертификатов в вашей компании могут накладываться ограничения, к примеру, на область действия сертификата или полей расширения. Мы столкнулись с тем, что у сертификатов после подписания была только возможность проверки подлинности сервера (serverAuth), а поле SAN необходимо было дополнительно задавать. При попытке запустить кластер на сертификатах с данным ограничением возникает ошибка.

javax.net.ssl|ERROR|19|/0.0.0.0:3888|2021-10-27 11:18:22.090 GMT|TransportContext.java:341|Fatal (CERTIFICATE_UNKNOWN): Extended key usage does not permit use for
TLS client authentication (
«throwable» : <
sun.security.validator.ValidatorException: Extended key usage does not permit use for TLS client authentication

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

Иногда стоит увеличить уровень логирования для более полного понимания картины при настройке и запуске, поэтому для получения более детальных логов в конфигурационном файле деплоя требуемого контейнера в параметр « KAFKA_OPTS » прописываем:

Аутентификация

Основные механизмы, поддерживаемые Kafka и предусмотренные протоколом SASL, для аутентификации по имени пользователя и паролю – PLAIN и SCRAM. PLAIN подразумевает передачу паролей в открытом виде, поэтому требует организации шифрования канала при реализации в промышленном контуре. В SCRAM пароли без криптографического преобразования не передаются, что повышает уровень защищенности, необходимость использования шифрования остается. В отличие от PLAIN, где пароли всех подключающихся пользователей хранятся в файлах JAAS на брокере, в SCRAM они записываются в хэшированном виде на серверы Zookeeper. Таким образом использование SCRAM добавляет бОльшие чем в PLAIN временные издержки на установление соединения, но в ходе дальнейшей передачи данных они нивелируются.

Рассмотрим настройку SASL PLAIN. Напомним, что при использовании SASL аутентификация клиента на уровне TLS не производится. В процессе настройки для каждого сервера конфигурируется JAAS файл с учетными данными и указанием способа аутентификации. При использовании SASL поверх SSL на брокерах мы указываем расположение JAAS файла внутри контейнера:

Внутри файла указываются учетные данные и механизм аутентификации внутри обособленных блоков. Блок описывает взаимодействие: к примеру, блок «KafkaServer» отражает данные для взаимодействия клиентов с брокерами. Блок «Client» описывает, под каким пользователем брокер обращается к кластеру Zookeeper. Пример JAAS файла:

KafkaServer <
org.apache.kafka.common.security.plain.PlainLoginModule required
username=»KafkAdmin»
password=»»
user_KafkaAdmin =»password»
user_User1=»password»;
>;

Client <
org.apache.kafka.common.security.plain.PlainLoginModule required
username=»ZooKafka»
password=»password»;
>;

На стороне Zookeeper при активации SASL задаются следующие параметры:

KAFKA_OPTS: «-Djava.security.auth.login.config=/etc/kafka/zookeeper_server_jaas.conf -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider -Dzookeeper.requireClientAuthScheme=sasl

В файле jaas указывается блок «Server», где описываются: пользователь, под которым взаимодействуют сами хосты Zookeeper, и пользователь, под которым брокеры обращаются к Zookeeper:

Server <
org.apache.kafka.common.security.plain.PlainLoginModule required
username=»ZooAdmin»
password=»pass»
user_ZooKafka=»pass»;
>;

При включении SASL на Zookeeper на созданные до этого Znode не распространяются применяемые правила, поэтому нужно выполнять zookeeper-security-migration.sh.

Авторизация

Для целей авторизации настроен ACL.

На стороне брокеров нужно выполнить ряд настроек:

Для ограничения возможности записи метаданных в Zookeeper задаем следующий параметр:

Настраивается суперпользователь, от которого брокеры будут работать в кластере:

JAAS файлы уже были представлены ранее как для брокеров, так и для Zookeeper — пользователям в них и задаются права ACL.

Всем приложениям, осуществляющим запись (producer) и чтение (consumer), заданы отдельные пользователи с разграничением минимально необходимых прав на манипуляции в кластере. Сonsumer в рамках приложений выделяются в группы (consumer group), пользователям выдаются права на чтение из своих групп.

Основные команды приведены по ссылке.

Инструменты

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

Prometheus JMX exporter

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

Для установки Prometheus JMX exporter нужно создать директорию для хранения его исполняемых и конфиг файлов:

Prometheus получает метрики с серверов Kafka за счет JMX exporter агента устанавливаемого на узлы кластера.

Скачиваем последнюю версию JMX exporter и настраиваем конфигурацию для работы на каждом брокере Kafka, например:

Загружаем конфигурационный файл последней версии для брокеров kafka:

Добавляем в .yml файл развертывания кластера kafka настройки для каждого брокера:

1) KAFKA_OPTS: «-javaagent:/etc/kafka/prometheus/jmx_prometheus_javaagent-0.16.1.jar=8088:/etc/kafka/prometheus/kafka-2_0_0.yml»

/etc/kafka/prometheus/jmx_prometheus_javaagent-0.16.1.jar –путь внутри контейнера до jar файла

8088 – порт, на котором будет слушаться JMXexporter (если кластер развернут на одном сервере, необходимо указывать разные порты для каждого брокера!)

/etc/kafka/prometheus/kafka-2_0_0.yml – путь внутри контейнера до конфигурационного файла.

Возможные ошибки, возникающие при использовании параметра -javaagent в блоке KAFKA_OPTS :

При использовании CLI kafka внутри контейнера возможно возникновение ошибок вида:

Exception in thread «main» java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at sun.instrument.InstrumentationImpl.loadClassAndStartAgent(InstrumentationImpl.java:386)
at sun.instrument.InstrumentationImpl.loadClassAndCallPremain(InstrumentationImpl.java:401)
Caused by: java.net.BindException: Address already in use
at sun.nio.ch.Net.bind0(Native Method)
at sun.nio.ch.Net.bind(Net.java:433)
at sun.nio.ch.Net.bind(Net.java:425)
at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
at sun.net.httpserver.ServerImpl.(ServerImpl.java:100)
at sun.net.httpserver.HttpServerImpl.(HttpServerImpl.java:50)
at sun.net.httpserver.DefaultHttpServerProvider.createHttpServer(DefaultHttpServerProvider.java:35)
at com.sun.net.httpserver.HttpServer.create(HttpServer.java:130)
at io.prometheus.jmx.shaded.io.prometheus.client.exporter.HTTPServer.(HTTPServer.java:190)
at io.prometheus.jmx.shaded.io.prometheus.jmx.JavaAgent.premain(JavaAgent.java:31)
. 6 more
FATAL ERROR in native method: processing of -javaagent failed
Aborted (core dumped)

Чтобы ошибка не возникала, нужно перед выполнением команд в CLI Kafka единоразово на всю сессию выполнить команду:

/app/docker/docker-images/prometheus – реальный путь на сервере

/etc/kafka/prometheus — путь внутри контейнера

Выполняем передеплой контейнеров kafka, используя команду:

sudo docker-compose -f /путь/к/файлу.yml up –d

Выполнить проверку генерации метрик JMXexporter-ом на хостах и портах указанных в настройках:

В конфиг – файле prometheus.yml на сервере с prometheus в блоке « scrape_configs » необходимо добавить строки вида:

Grafana

Для визуализации получаемых метрик можно использовать Grafana, в ней удобно настроить свои dashboards или скачать готовые решения. Для сбора метрик нужно установить и настроить сборщик Prometheus. В качестве альтернативы можно использовать Zabbix, он прекрасно работает с JMX метриками и может брать их напрямую с хоста экспортера без Prometheus.

Instana

Для мониторинга кластеров Kafka можно использовать систему Instana с автодискаверингом. При работе с ней были выявлены некоторые ньюасы, о которых мы расскажем ниже.

Помимо штатных показателей производительности кластера kafka и встроенных детекторов аномального состояния кластера, у Instana есть очень полезная функция — отображение Consumer Groups Lag Per Topic. Она подсчитывает лаги Consumer групп в разрезе топиков, визуализирует данные в графическом виде, позволяет взглянуть на них в реальном времени или во временных интервалах. В качестве альтернативы отображения Consumer Groups Lag можно использовать решение от linkedin — Burrow, он имеет встроенный Prometheus exporter (начиная с версии 1.3.4) и хорошо визуализирует лаги работая вкупе со стеком Prometheus + Grafana.

Техподдержка Instana утверждает следующее: «We use the Kafka broker API in order to calculate lag as a diff between last committed and last consumed offset for specific consumer group and topic».

Для отображения Consumer Groups Lag Per Topic агентом Instana используется следующая команда вида:

kafka-consumer-groups —describe —all-groups —bootstrap-server localhost:9092

Внимательные читатели могли заметить, что в кластере kafka с настроенными ACL листами выполнение CLI команды (на примере кластера на основе дистрибутива от confluent) docker exec -it container_name kafka-consumer-groups —describe —all-groups —bootstrap-server kafka1.test:9092 совершенно справедливо будет возвращать пустой результат или выдавать ошибку, причин такому поведению несколько:

команда по умолчанию обращается к кластеру Kafka по протоколу PLAINTEXT, соответственно, она работает исключительно при настройке кластера на работу с данным протоколом (параметр KAFKA_ADVERTISED_LISTENERS );

при настроенном параметре KAFKA_ADVERTISED_LISTENERS PLAINTEXT:// при запросе будет выдаваться пустой результат, т.к. на кластере включен ACL, а по протоколу PLAINTEXT, как мы рассматривали выше, осуществляется анонимный доступ, без указания учетных данных;

при отсутствии настроенного параметра KAFKA_ADVERTISED_LISTENERS PLAINTEXT:// возникнет ошибка Error: Executing consumer group command failed due to org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups ;

для обращения по другим протоколам нужно использовать команду вида:
docker exec -it container_name kafka-consumer-groups —describe —all-groups —bootstrap-server kafka1.test:9096 —command-config /путь/до/jaas-конфиг.conf ;

где jaas-конфиг.conf это файл со строками подключения по протоколу SASL_PLAINTEXT или SASL_SSL и логином/паролем с необходимыми правами.

Примеры таких конфигов:

Конфиг для работы по протоколу SASL_PLAINTEXT:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
username=»user»
password=»password»;

Конфиг для работы по протоколу SASL_SSL:

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
ssl.truststore.location=/etc/kafka/secrets/kafka.truststore.jks
ssl.truststore.password=password
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
username=»user»
password=»password»;

Нужно обратить внимание на использование отличных от задаваемых по умолчанию (9092) портов, если путь до server.properties брокеров не прописан в конфигурации агента Instana, он будет искать конфиг файлы в следующих путях:

Аргументы процесса брокера Kafka.

Переменная окружения KAFKA_SERVER_PROPERTIES

Агент Instana использует путь /opt/kafka/config/server.properties по умолчанию, если путь к server.properties не удалось найти ни одним из вышеупомянутых способов.

Не найдя конфиг файлы по данным путям, агент Instana будет стучаться до кластера Kafka по порту по умолчанию – 9092, что приведет к многочисленным ошибкам подключения к кластеру Kafka и отсутствию некоторых данных в вэб консоли Instana.

Немного о возникших проблемах при использовании Instana. Механизм подсчета Consumer Groups Lag Per Topic иногда сбоит и начинает показывать значение, кратное количеству брокеров, т.е. при реальном лаге в 300 сообщений, на 3х брокерной конфигурации возможно отображение лага в 1500 сообщений. Также данную фичу сейчас нельзя использовать на кластерах с настроенными ACL, т.к. отдельной настройки под механизм Consumer Groups Lag Per Topic в Instana нет. Мы завели в компании Instana Feature Request для добавления функционала тонкой настройки механизма Consumer Groups Lag Per Topic и ожидаем его добавления в ближайшем будущем.

Kafka Cruise Control + Cruise Control UI

мониторинг состояния брокеров, топиков и партиций;

мониторинг нагрузки и утилизации ресурсов кластера;

продвинутую балансировку нагрузки;

детектирование аномалий и автохилинг;

автоматизацию добавления/удаления брокеров.

Kafka GitOps

Еще один незаменимый инструмент — Kafka GitOps. Позволяет автоматизировать управление топиками и списками управления доступом (ACL) Apache Kafka.

Топики и ACL определяются и ведутся в файле YAML формата. Идея проекта — управление топиками и ACL Kafka через версионные файлы конфигурации. При запуске Kafka GitOps сравнивает желаемое состояние кластера Kafka с фактическим и создает план для выполнения в кластере. После выполнения топики и ACL будут соответствовать желаемому состоянию.

Kafka GitOps также создает необходимые ACL для каждого типа приложений. Нет необходимости вручную создавать кучу списков ACL для Kafka Connect, Kafka Streams и т. д.

Работа с консолью:

По дефолту приложение ищет файл state.yaml в текущей директории. Если конфиг лежит в той же директории, что и jar-файл, то параметр -f при вызове указывать не нужно.

Примеры стандартных команд запуска приложения из консоли:

java -jar kafka-gitops-all.jar -f /kafka/statefile/topics.yaml

java -jar kafka-gitops-all.jar

Ниже приведены аргументы (доступны по параметру -h )

-f, —file= расположение файла с конфигом
—no-delete только изменять и добавлять, не удалять топики\ACL-листы
—skip-acls не анализировать ACL-листы из конфига, настраивать только топики Kafka
-v, —verbose детальный вывод в консоль
-o расположение, куда записать получившийся план (если мы хотим только сформировать план, не применяя изменения к кластеру Kafka)

Команды:

apply применить конфиг к Kafka
plan только построить план изменений, не применяя к Kafka
validate валидировать конфиг на предмет наличия\отсутствия ошибок в state.yaml файле

Пример вызова приложения:

java -jar kafka-gitops-all.jar -f /kafka/statefile/topics.yaml —skip-acls —no-delete —verbose plan

тут мы запускаем программу, говорим ей не анализировать ACL-листы (—skip-acls );
не пытаться удалить, а только изменить и добавить топики\ACl-листы ( —no-delete );
выводить все промежуточные шаги в консоль (—verbose ), только формировать план изменений ( plan ), так как мы не добавляли флаг -o , то план изменений будет только выведен в консоль и не будет сохранен в файл).

Конфиг (файл state.yaml)

Блок общих настроек:

settings:
topics:
defaults:
replication: 1 # (ко всем топикам из конфига по умолчанию применять фактор репликации = 1, если не указано иного)
blacklist:
prefixed:
— _confluent # (игнорировать уже имеющиеся в Kafka топики, название которых начинается на _confluent)

Блок настроек топиков:

topics:
msg: # (название топика)
partitions: 3 # (кол-во партиций)
replication: 1 # (фактор репликации)
configs:
cleanup.policy: compact # (любые параметры настройки топиков)
segment.bytes: 1000000 # (любые параметры настройки топиков)
test-msg:
partitions: 2
replication: 1
configs:
cleanup.policy: compact
segment.bytes: 1000000

Переменные окружения:

Параметры подключения к кафке настраиваются через параметры окружения:

KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_SASL_JAAS_PASSWORD=kafka
KAFKA_SASL_JAAS_USERNAME=kafka
KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT
KAFKA_CLIENT_ID=gitops-user
KAFKA_SASL_MECHANISM=PLAIN

Если на кластере Apache Kafka нет никакой авторизации, то достаточно будет прописать параметр KAFKA_BOOTSTRAP_SERVERS

Удобный и функциональный GUI менеджер для Apache Kafka — AKHQ.

Его ключевые особенности:

позволяет организовать мультикластерную конфигурацию из единого интерфейса;

поддержка протоколов и механизмов аутентификации и шифрования для Apache Kafka;

управление топиками, consumer группами, конфигурациями кластера, пользователями, ACL и т.д.;

интеграция с LDAP, RBAC, Schema Registry, Kafka Connect и многими другими продуктами и технологиями;

тонкая настройка прав доступа к web-интерфейсу системы.

Пример .yml конфиг файла для быстрого старта с использованием docker контейнера:

version: ‘3.6’
services:
akhq:
image: tchiotludo/akhq
environment:
AKHQ_CONFIGURATION: |
akhq:
connections:
cluster-1:
properties:
bootstrap.servers: «server1:19092, server2:29092, server3:39092»
cluster-2:
properties:
bootstrap.servers: » server1:19092, server2:29092, server3:39092″
cluster-3:
properties:
bootstrap.servers: » server1:19092, server2:29092, server3:39092″
security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username=»user» password=»password»;
network_mode: host

Offset explorer

Offset Explorer (ранее Kafka Tool) — толстый клиент для управления и использования кластеров Apache Kafka с поддежкой мультикластерности. Он позволяет быстро просматривать объекты в кластере Kafka, а также сообщения, хранящиеся в топиках кластера. Offset Explorer содержит функционал, ориентированный как на разработчиков, так и на администраторов.

Некоторые из ключевых особенностей:

быстрый просмотр кластеров Kafka, включая их брокеров, топики и consumer’ов;

просмотр содержимого и добавление новых сообщений в партициях;

просмотр и выставление offset у consumer;

вывод сообщений в JSON, XML и Avro формате с возможностью их локального сохранения;

управление топиками, их конфигурациями и ACL листами;

поддержка протоколов и механизмов аутентификации и шифрования для Apache Kafka;

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

Вывод

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

Kafka Search

Every article I wrote mentioned that Kafka is everywhere in Trendyol. Trendyol has many tech stacks. Every backend team has used many services but the most important one is Kafka.

Every time, Searching has been an issue. Nowadays, there are some solutions to searching in Kafka. First, let's look into them.

Every Event transfer to ElasticSearch:

You can transfer events via the Kafka connector.
It looks like the better option at first view but, you have to maintain a new big service. Also, you allocate some resources of Cluster and you have to make sure every Event upserted into ElasticSearch
When you can’t find an event in Elastic. You will ask this question yourself in doubt, "what if it wasn’t delivered?"

Offset Explorer:

You can install your local machine and use it if you don’t have any access issues to the production environment. Also, it’s too slow.
In my experience, it takes about 4 minutes to get search results from it, and freezes at every step.

Kowl or something:

You can use it for searching but, when you start searching for a Topic with huge messages, they cannot respond to users many times.
In my experience, it takes under a minute to get search results from it. During the get result, they crashed 1 time.

What is the problem?

If the topic has a high message count or more than one user uses the same time, the apps I mentioned above or method run slowly. Sometimes they can't respond to UI and crash because searching for any word in the Topic which has 1 billion messages is a long-term process and it's CPU bound process. It consumes every CPU resource.

Another problem is where is the Topic? Trendyol uses more than one cluster. When we search the word in the Topic, we have to know which Cluster has the Topic otherwise, we can't search. When you use the apps I mentioned above have to search Cluster by Cluster.

What do we try to achieve here?

I developed a macro service for Event searching in Kafka. You can search any word in key or value at the specified topic without any freeze, crash, or slowness.

What is the difference here?

Here, we separate the searching process and the serving process. Searching is a really massive consuming CPU resources. Two applications (API/searching consumer) are working with Kafka and Redis.
When we request to API to start searching, we pass the process to the searching consumer. We don't use resources of API during the search. Therefore, the search process doesn't affect the API.

How it works:

I sent this request http://localhost:8080/search?topicName=product.spda.enricher.invalidateListing.1&value=843078897 and refreshed the page to get results in seconds.

Api and consumer share search data via Redis. When the consumer starts searching for the value in the Topic and finds data then saves the data to the Redis. Api gets data from Redis and presents it to the user.

1. Определение

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

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

Три общие семантики в системах потоковой обработки:

Максимум однажды	Каждая запись обрабатывается один раз или не обрабатывается совсем
Хотя бы один раз	Это лучше, чем один раз, потому что это гарантирует, что никакие данные не будут потеряны. Но могут быть дубликаты
Да и только один раз	Каждая запись будет обработана ровно один раз, данные не будут потеряны, и никакие данные не будут обрабатываться несколько раз.

At most once: Each record will be either processed once or not processed at all.
At least once: Each record will be processed one or more times. This is stronger than at-most once as it ensure that no data will be lost. But there may be duplicates.
Exactly once: Each record will be processed exactly once — no data will be lost and no data will be processed multiple times. This is obviously the strongest guarantee of the three.

2.Kafka offset Management with Spark Streaming

В первую очередь рекомендуется сохранять смещение в Zookeeper. Zookeeper более легкий, чем HBASE и т. Д., И он используется для HA (высокая доступность), а смещение более безопасно.

Стандартные двухэтапные операции для управления смещением:

Сохранить смещения
Получить компенсацию

3. Экологическая подготовка

Запустите продюсер Kafka и протестируйте тему: tp_kafka:

Запустите потребителя Kafka:

Создавайте данные в IDEA:

4. Первый метод управления смещением: наименьшее

Количество Spark Streaming link статистика Kafka:

Воспроизвести 100 данных Kafka -> Spark Streaming принимает:

Но если Spark Streaming останавливается и перезапускается в это время:

Вы обнаружите, что отсчет начинается здесь снова, потому что значение auto.offset.reset установлено на наименьшее значение в коде. (До версии kafka-0.10.1.X)

5. Второй способ управления смещением: контрольно-пропускной пункт

Создайте папку / offset в HDFS:

Использовать контрольную точку:

Примечание. IDEA изменяет пользователя HDFS в параметрах виртуальной машины в настройках:

Установил, что предыдущие 100 штук были израсходованы. Это после остановки, производства 100 штук, а затем запуска:

Было обнаружено, что здесь было прочитано только 100 элементов между концом последнего раза и началом этого времени, вместо того, чтобы читать все предыдущие элементы, как самые маленькие.

Но у checkpiont есть проблема: если смещение управляется таким образом, пока бизнес-логика меняется, контрольная точка не действует. Потому что он вызывает getOrCreate ().

6. Третий тип управления смещением: ручное управление смещением.

Создать StreamingContext
Получить данные от Kafka <== получить смещение
Процесс согласно бизнес-логике
Записать результат обработки во внешнее хранилище ==> сохранить смещение
Запустите программу и дождитесь завершения потока

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

Решение 1: реализовать идемпотент

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

Решение 2: транзакция (транзакция)

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

2. Либо все эти операции с базой данных, составляющие логическое целое, выполняются успешно, либо все не выполняются.

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

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

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