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

Делаем свой десктопный GUI к Apache Kafka или Conduktor для обездоленных

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

Как-то так сложилось, что у нас не так много UI для Apache Kafka. А если хочется именно desktop, то Offset Explorer и упомянутый Conduktor. Первый имеет морально устаревший интерфейс 2000х, а второй не оправдано дорогой, т. к. не использую весь его богатый функционал. Вооружившись Qt и librdkafka, набросал conduktor на минималках.

Используй layout Люк

Сложный выпадающий элемент, имеющий множество состояний. Как съесть слона? По кусочкам маленького размера. Лайауты умеют отслеживать изменение видимости компонентов, и высчитывать размеры на основе implicit size объекта. Более мелки компоненты строятся следующим образом

Item < id: item implicitHeight: layout.implicitHeight implicitWidth: layout.implicitWidth property int selectedLimit: 0 ColumnLayout < id: layout SpinBox < visible: item.selectedLimit == 1 >TextField < visible: item.selectedLimit == 2 >SpinBox < visible: item.selectedLimit == 3 >> >

Сворачивающая левая панель построена на манипуляции с размерами

states: [ State < name: "default" PropertyChanges < target: collapsBtnLabel text: qsTr("« Collapse") >PropertyChanges < target: menu width: 300 implicitWidth: 300 >PropertyChanges < target: header state: "default" >PropertyChanges < target: kafka_icon source: "qrc:/kafka_icon.svg" >>, State < name: "small" PropertyChanges < target: collapsBtnLabel text: "»" >PropertyChanges < target: menu width: 60 implicitWidth: 60 >PropertyChanges < target: header state: "small" >PropertyChanges < target: kafka_icon source: menu.broker.color !== Style.BrokerColor[0] ? "qrc:/kafka_icon_reverse.svg" : "qrc:/kafka_icon.svg" >> ]

И для сравнения как сделано в Qt Quick Controls

T.CheckDelegate < id: control implicitWidth: Math.max(implicitBackgroundWidth + leftInset + rightInset, implicitContentWidth + leftPadding + rightPadding) implicitHeight: Math.max(implicitBackgroundHeight + topInset + bottomInset, implicitContentHeight + topPadding + bottomPadding, implicitIndicatorHeight + topPadding + bottomPadding) padding: 6 spacing: 6 icon.width: 16 icon.height: 16 contentItem: IconLabel < leftPadding: control.mirrored ? control.indicator.width + control.spacing : 0 rightPadding: !control.mirrored ? control.indicator.width + control.spacing : 0 spacing: control.spacing mirrored: control.mirrored display: control.display alignment: control.display === IconLabel.IconOnly || control.display === IconLabel.TextUnderIcon ? Qt.AlignCenter : Qt.AlignLeft icon: control.icon text: control.text font: control.font color: control.highlighted ? Fusion.highlightedText(control.palette) : control.palette.text >indicator: CheckIndicator < x: control.mirrored ? control.leftPadding : control.width - width - control.rightPadding y: control.topPadding + (control.availableHeight - height) / 2 control: control >background: Rectangle < implicitWidth: 100 implicitHeight: 20 color: control.down ? Fusion.buttonColor(control.palette, false, true, true) : control.highlighted ? Fusion.highlight(control.palette) : control.palette.base >> 

Twist таблицы и списка

Твист заключается в переключении между двумя режимами отображения

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

Rectangle < property int rowHeight: 40 StackLayout < anchors.fill: parent ListView < Layout.fillWidth: true Layout.fillHeight: true ScrollBar.vertical: ScrollBar < id: listVerticalBar policy: ScrollBar.AsNeeded minimumSize: 0.06 onPositionChanged: tableVerticalBar.position = position >delegate: Rectangle < implicitHeight: rowHeight >> TableView < Layout.fillWidth: true Layout.fillHeight: true ScrollBar.vertical: ScrollBar < id: tableVerticalBar policy: ScrollBar.AsNeeded minimumSize: 0.06 onPositionChanged: listVerticalBar.position = position >delegate: Item < implicitHeight: rowHeight >> > >

Делегат для таблицы выглядит следующим образом

Item < implicitWidth: 100 implicitHeight: rowHeight StackLayout < anchors.fill: parent currentIndex: column Text < // topic >Text < // part >//. > >

Нехитрая схема, которая позволяет настраивать вид каждой колонки. Если не боитесь дополнительных зависимостей, то можете взять DelegateChooser и DelegateChoice.

Ещё раз о таблицах

Изменение количества колонок в таблице и их размер, это привычный функционал и что-то само собой разумеется. Что может пойти не так? TableView позволяет задавать функцию columnWidthProvider , которая позволяет устанавливать ширину столбца. Начиная с Qt 5.13 если вернуть 0, колонка будет скрыта.

Что бы применить изменения, дергаем forceLayout , как это сделано в документации

TableView < id: tableView property var columnWidths: [100, 50, 80, 150] columnWidthProvider: function (column) < return columnWidths[column] >Timer < running: true interval: 2000 onTriggered: < tableView.columnWidths[2] = 150 tableView.forceLayout(); >> >

В моём случае это выглядит так

Rectangle < id: main property var columnVisible: [true, true, true, true, true, true, true] property var columnWidths: [100, 50, 100, 150, 100, 350, 200] function columnWidthProvider(column) < let visible = columnVisible[column]; let width = visible ? columnWidths[column] : 0; return width; >function hideColumn(column, hide) < columnVisible[column] = hide; view.forceLayout(); >//. TableView < id: view columnWidthProvider: main.columnWidthProvider >>

Осталось за малым, вывести заголовок таблицы и сделать изменение размера столбца. Естественно нашелся компонент под это дело, а именно HorizontalHeaderView

HorizontalHeaderView < id: horizontalHeader reuseItems: false syncView: view height: 30 Layout.fillWidth: true delegate: Rectangle < id: root implicitWidth: 50 implicitHeight: 30 Text < anchors.centerIn: parent text: display color: Style.LabelColor font.bold: true >Rectangle < id: splitter color: Style.BorderColor height: parent.height width: 1 visible: mouseArea.containsMouse x: columnWidths[index] - 1 onXChanged: < if (drag.active) < main.columnWidths[index] = splitter.x + 1; view.forceLayout(); >> DragHandler < id: drag yAxis.enabled: false xAxis.enabled: true cursorShape: Qt.SizeHorCursor >> > >

Да, это работает. Беремся за splitter и двигаем влево, вправо. Какие тут подводные камни? Понимающие люди обратили внимание на reuseItems: false . HorizontalHeaderView это view, а в view используется пул элементов(reusing items), что бы сэкономить на создание и удалении. Документация не рекомендует иметь делегаты с состоянием.

На что влияет reuseItems: true в данном примере. Представьте, вы взялись за splitter и растягиваете колонку, которая имеет ширину 300. В какой-то момент view решает пере использовать элемент, возвращает в пул, достаёт от туда, и вставляет в другое место и инициализирует шириной 40. Получаем не очень понятное поведение. В моём случае такое поведение проявляется при вставке/удалении строк таблицы.

Про окна

Интерфейс не перегружен окнами как MDI и не является SDI . Окна создаются динамически, что бы меньше имели общего состояния. Каждый такой компонент Window и создается следующим образом

function createConsumerScreen(topic, topicModel, broker) < let component = Qt.createComponent("qrc:/qml/Consumer/ConsumerScreen.qml"); let posX = appWindow.x + appWindow.width/2 - Constants.ConsumerScreenWidth/2; let posY = appWindow.y + appWindow.height/2 - Constants.ConsumerScreenHeight/2; let state = < x: posX, y: posY, topic:topic, topicModel: topicModel, broker: broker >; let consumer = component.createObject(appWindow, state); > function createMessageScreen(x,y,width,height, message) < let component = Qt.createComponent("qrc:/qml/Consumer/MessageView.qml"); let posX = x + width/2 - Constants.MessageViewWidth/2; let posY = y + height/2 - Constants.MessageViewHeight/2; let state = < x: posX, y: posY, message:message, >; let consumer = component.createObject(appWindow, state); >

Заключение

Получился прототип, который можно развивать до полноценного MVP. Из ближайших планов настроить какой-нибудь CI для сборок под Windows и Mac OS X. Весь код доступен на Git Hub.

P.S.

Передаю пламенный привет Антону Водостоеву из 2ГИС

The Ultimate UI Tool for Kafka

Offset Explorer (formerly Kafka Tool) is a GUI application for managing and using Apache Kafka ® clusters. It provides an intuitive UI that allows one to quickly view objects within a Kafka cluster as well as the messages stored in the topics of the cluster. It contains features geared towards both developers and administrators. Some of the key features include

• Quickly view all your Kafka clusters, including their brokers, topics and consumers
• View contents of messages in your partitions and add new messages
• View offsets of the consumers, including Apache Storm Kafka spout consumers
• Show JSON, XML and Avro messages in a pretty-printed format
• Add and drop topics plus other management features
• Save individual messages from your partitions to local hard drive
• Write your own plugins that allow you to view custom data formats
• Offset Explorer runs on Windows, Linux and Mac OS

Offset Explorer is free for personal use only. Any non-personal use, including commercial, educational and non-profit work is not permitted without purchasing a license. Non-personal use is allowed for evaluation purposes for 30 days following the download of Offset Explorer, after which you must purchase a valid license or remove the software.

UI Tool for Apache Kafka

© 2015-2024 DB Solo, LLC. All rights reserved.
Apache Kafka is a trademark of the Apache Software Foundation. Offset Explorer is not produced by or affiliated with the Apache Software Foundation.

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.

SASL

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.

SSL

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.

Azure Event Hubs

In order to connect Offset Explorer to an Azure Event Hub, you must first make sure that «Kafka Surfaces» is enabled. This can be confirmed by navigating to the Event Hubs namespace page and seeing if it is enabled.

Once this has been confirmed, navigate to the «Shared access policies» menu item in the Event Hubs namespace. You can either use the RootManageSharedAccessKey, or make your own policy. Once you have chosen the policy, click on it and copy the connection string–primary key value.

Once the link has been copied, you can open Offset Explorer and add a new connection. In the properties tab, choose a name and version for the cluster. Note that Azure Event Hubs do not support access to Zookeeper so you must leave the Zookeeper host field empty.

In the security tab, change the security type from plaintext to SASL SSL.

In the advanced tab, make sure to use port number 9093, as it differs from the standard port for Kafka (9092). You will also need to insert PLAIN into the SASL mechanism field.

For the JAAS Config tab, you will need to enter the following information (note that the password is equal to the connection string–primary key link that you copied earlier):

Once all this information has been entered, you can click «Add» to add the connection.

Amazon MSK

The easiest method to connect to your Amazon MSK cluster is to configure your MSK security settings to allow SASL/SCRAM authentication. In order to connect Offset Explorer to your Amazon MSK environment using SASL/SCRAM authentication, you will need to add the following settings when creating your cluster in Offset Explorer:

Properties tab: Confirm that zookeeper host is left blank. Pick the version number that matches your Amazon MSK cluster.

Security tab: Confirm that the broker security type is SASL SSL.

Advanced tab: Confirm the bootstrap servers input is correctly given. Also, enter the following into the SASL Mechanism box: SCRAM-SHA-512

JAAS Config tab: Copy the following, switching out «myuser» for correct username and «mypassword» for correct password. These credentials are stored in the AWS Secrets Manager secret that is associated with your MSK SASL/SCRAM authentication.

Once these settings for the cluster have been given, you can add the server connection by clicking «Add» and connect.

Confluent Cloud

In order to connect Offset Explorer to a Confluent Cloud environment, you will need to add the following settings when creating your cluster in Offset Explorer:

Properties tab: Confirm that zookeeper host is left blank.

Security tab: Confirm that the broker security type is SASL SSL.

Advanced tab: Confirm the bootstrap servers input is correctly given. Also, enter the following into the SASL Mechanism box: PLAIN

JAAS Config tab: Copy the following, switching out «myapikey» and «myapisecret» for correct values.

Once these settings for the cluster have been given, you can add the cluster by clicking «Add» and connect.

Offset Explorer | UI Tool for Apache Kafka

© 2015-2023 DB Solo, LLC. All rights reserved.

Apache Kafka: основы технологии

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

Статья подготовлена на основе открытого занятия из видеокурса по Apache Kafka. Авторы — Анатолий Солдатов, Lead Engineer в Авито, и Александр Миронов, Infrastructure Engineer в Stripe. Базовые темы курса доступны на Youtube.

Базовые компоненты классической системы очередей

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

Консьюмеры получают данные с сервера, используя две разные модели запросов: pull или push.

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

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

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

Типичный жизненный цикл сообщений в системах очередей:

1. Продюсер отправляет сообщение на сервер.
2. Консьюмер фетчит (от англ. fetch — принести) сообщение и его уникальный идентификатор сервера.
3. Сервер помечает сообщение как in-flight. Сообщения в таком состоянии всё ещё хранятся на сервере, но временно не доставляются другим консьюмерам. Таймаут этого состояния контролируется специальной настройкой.
4. Консьюмер обрабатывает сообщение, следуя бизнес-логике. Затем отправляет ack или nack-запрос обратно на сервер, используя уникальный идентификатор, полученный ранее — тем самым либо подтверждая успешную обработку сообщения, либо сигнализируя об ошибке.
5. В случае успеха сообщение удаляется с сервера навсегда. В случае ошибки или таймаута состояния in-flight сообщение доставляется консьюмеру для повторной обработки.

Типичный жизненный цикл сообщений в системах очередей

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

Как и сервисы обработки очередей, Kafka условно состоит из трёх компонентов:

1) сервер (по-другому ещё называется брокер),
2) продюсеры — они отправляют сообщения брокеру,
3) консьюмеры — считывают эти сообщения, используя модель pull.

Базовые компоненты Kafka

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

• Сообщения в Kafka не удаляются брокерами по мере их обработки консьюмерами — данные в Kafka могут храниться днями, неделями, годами.
• Благодаря этому одно и то же сообщение может быть обработано сколько угодно раз разными консьюмерами и в разных контекстах.

Теперь давайте посмотрим, как Kafka и системы очередей решают одну и ту же задачу. Начнём с системы очередей.

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

1) отправить письмо пользователю,
2) пересчитать дневную статистику регистраций.

В случае с RabbitMQ или Amazon SQS функционал может помочь нам доставить сообщения всем сервисам одновременно. Но при необходимости подключения нового сервиса придётся конфигурировать новую очередь.

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

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

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

Структура данных

Наверняка возникает вопрос: «Раз сообщения не удаляются, то как тогда гарантировать, что консьюмер не будет читать одни и те же сообщения (например, при перезапуске)?».

Для ответа на этот вопрос разберёмся, какова внутренняя структура Kafka и как в ней хранятся сообщения.

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

Например:

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

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

Для гарантии сохранности данных каждая партиция в Kafka может быть реплицирована n раз, где n — replication factor. Таким образом гарантируется наличие нескольких копий сообщения, хранящихся на разных брокерах.

У каждой партиции есть «лидер» (Leader) — брокер, который работает с клиентами. Именно лидер работает с продюсерами и в общем случае отдаёт сообщения консьюмерам. К лидеру осуществляют запросы фолловеры (Follower) — брокеры, которые хранят реплику всех данных партиций. Сообщения всегда отправляются лидеру и, в общем случае, читаются с лидера.

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

Основная структура данных в Kafka — это распределённый, реплицируемый лог. Каждая партиция — это и есть тот самый реплицируемый лог, который хранится на диске. Каждое новое сообщение, отправленное продюсером в партицию, сохраняется в «голову» этого лога и получает свой уникальный, монотонно возрастающий offset (64-битное число, которое назначается самим брокером).

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

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

Consumer Groups

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

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

Вернёмся к нашему примеру с топиком сервиса регистрации и представим, что у сервиса отправки писем есть своя собственная консьюмер-группа с одним консьюмером
c1
внутри. Значит, этот консьюмер будет получать сообщения из всех партиций топика.

Если мы добавим ещё одного консьюмера в группу, то партиции автоматически распределятся между ними, и
c1
теперь будет читать сообщения из первой и второй партиции, а
c2
— из третьей. Добавив ещё одного консьюмера (c3), мы добьёмся идеального распределения нагрузки, и каждый из консьюмеров в этой группе будет читать данные из одной партиции.

А вот если мы добавим в группу ещё одного консьюмера (c4), то он не будет задействован в обработке сообщений вообще.

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

Если консьюмеры не справляются с текущим объёмом данных, то следует добавить новую партицию в топик. Только после этого консьюмер c4 начнёт свою работу.

Механизм партиционирования является нашим основным инструментом масштабирования Kafka. Группы являются инструментом отказоустойчивости.
Кстати, как вы думаете, что будет, если один из консьюмеров в группе упадёт? Совершенно верно: партиции автоматически распределятся между оставшимися консьюмерами в этой группе.

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

1. Гарантия очерёдности данных — если вы пишете сообщения с ключами и хешируете номер партиции для сообщений, исходя из общего числа, то при добавлении новой партиции вы можете просто сломать порядок этой записи.
2. Партиции невозможно удалить после их создания, можно удалить только весь топик целиком.

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

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

А теперь, зная внутреннее устройство консьюмеров в Kafka, давайте вернёмся к изначальному вопросу: «Каким образом мы можем обозначить сообщения в партиции, как обработанные?».

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

Консьюмер делает специальный запрос к брокеру, так называемый offset-commit с указанием своей группы, идентификатора топик-партиции и, собственно, офсета, который должен быть отмечен как обработанный. Брокер сохраняет эту информацию в своём собственном специальном топике. При рестарте консьюмер запрашивает у сервера последний закоммиченный офсет для нужной топик-партиции, и просто продолжает чтение сообщений с этой позиции.

В примере консьюмер в группе email-service-group, читающий партицию p1 в топике registrations, успешно обработал три сообщения с офсетами 0, 1 и 2. Для сохранения позиций консьюмер делает запрос к брокеру, коммитя офсет 3. В случае рестарта консьюмер запросит свою последнюю закоммиченную позицию у брокера и получит в ответе 3. После чего начнёт читать данные с этого офсета.

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

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

Представим, что обработка сообщения с офсетом 1 завершилась с ошибкой. Однако мы продолжили выполнение нашей программы в консьюмере и запроцессили сообщение с офсетом 2 успешно. В таком случае перед нами будет стоять выбор: какой офсет закоммитить — 1 или 3. В настоящей системе мы бы рекомендовали закоммитить офсет 3, добавив при этом функционал, отправляющий ошибочное сообщение в отдельный топик для повторной обработки (ручной или автоматической). Подобные алгоритмы называются Dead letter queue.

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

Apache ZooKeeper

В заключение нужно упомянуть об ещё одном важном компоненте кластера Kafka — Apache ZooKeeper.

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

В случае падения брокера именно в ZooKeeper контроллером будет записана информация о новых лидерах партиций. Причём с версии 1.1.0 это будет сделано асинхронно, и это важно с точки зрения скорости восстановления кластера. Самый простой способ превратить данные в тыкву — потеря информации в ZooKeeper. Тогда понять, что и откуда нужно читать, будет очень сложно.

В настоящее время ведутся активные работы по избавлению Kafka от зависимости в виде ZooKeeper, но пока он всё ещё с нами (если интересно, посмотрите на Kafka improvement proposal 500, там подробно расписан план избавления от ZooKeeper).

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

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