Введение
Группа может распределять разделы топика между несколькими потребителями. Внутри группы со стабильным назначением у каждого раздела один активный владелец, а другая группа может независимо читать те же сохранённые записи. Это позволяет распределять работу без конкуренции всех приложений за единую общую позицию чтения.
Вы запустите два длительно работающих консольных клиента в одной группе, изучите их фактические назначения до публикации и сравните результаты чтения отдельной группы. После выхода одного участника вы дождётесь, когда оставшийся участник получит оба раздела, и только затем отправите ещё один небольшой пакет. Подготовленный инструмент публикации явно направляет записи в оба раздела, чтобы в эксперименте у каждого была работа; фактическая доставка на брокер проверяется.
В этой независимой работе предоставлены топик с двумя разделами, доступные для просмотра учебные события и инструменты ожидания с ограниченным временем работы, которые только читают состояние. Достаточно базовых навыков работы в терминале; инструменты для учебных данных не требуют редактирования Python. Консольные клиенты работают в фоне, пока вы не остановите их на заключительных шагах. Оставайтесь в той же оболочке Terminal, чтобы переменные с идентификаторами их процессов оставались доступными.
Запустите двух участников и дождитесь стабильного назначения
На этом шаге вы изучите назначение разделов до передачи работы группе. В подготовленном топике два раздела и пока нет записей. Просмотрите его метаданные:
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
Следующие клиенты используют одну группу orders-team. Параметр --command-property передаёт каждое значение конфигурации клиенту потребителя. client.id обозначает их в отчётах как reader-a и reader-b. Знакомые настройки автоматической фиксации позволяют позиции чтения стабилизироваться, пока клиенты продолжают опрос.
Символ & в конце запускает клиент в фоне оболочки, позволяя выполнять команды просмотра в том же терминале. $! — идентификатор только что запущенного фонового процесса; каждое присваивание сразу после запуска сохраняет его для последующей остановки именно этого процесса.
Оболочка поддерживает два потока вывода: стандартный вывод для записей и стандартный поток ошибок для диагностики. > записывает первый поток в файл JSONL; 2> записывает поток ошибок в журнал. Присваивание consumer_a_pid=$! сохраняет переменную оболочки без пробелов вокруг =; "$consumer_a_pid" позднее извлекает именно этот идентификатор процесса. Обычные записи и диагностика попадают в отдельные файлы, чтобы не мешать работе в командной строке. Эти клиенты работают до команд kill в конце, которые останавливают конкретные процессы:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-a --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-a.jsonl 2> /home/labex/project/output/reader-a.log &
consumer_a_pid=$!
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-b --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-b.jsonl 2> /home/labex/project/output/reader-b.log &
consumer_b_pid=$!
Дождитесь фактического назначения для двух участников и сохраните отчёт:
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
Вспомогательный инструмент только читает состояние, повторно запрашивая отчёт об участниках группы, пока два обозначенных клиента не получат непустые непересекающиеся назначения, которые вместе охватывают разделы 0 и 1. Время ожидания ограничено; инструмент не создаёт вымышленные назначения и не меняет позицию чтения.
В отчёте CLIENT-ID обозначает reader-a или reader-b, а CURRENT-ASSIGNMENT перечисляет фактические текущие назначения. Например, orders-shared:0 означает раздел 0 этого топика. Целевое назначение ещё может ожидать применения, поэтому смотрите текущий столбец. Ожидается, что каждый участник получит один раздел, но reader-a не обязан получить конкретный номер раздела. Перебалансировка перераспределяет разделы при присоединении или выходе участников группы. Группе нужно время, чтобы прийти к новому назначению; запуск процесса не означает, что владение разделами уже установилось. Стабильное назначение подтверждает распределение работы; сами по себе два идентификатора процессов не доказывают, какими разделами они владеют.
Опубликуйте данные для обработки и сравните с независимой группой
На этом шаге вы передадите записи в оба назначенных раздела и понаблюдаете, как вторая группа читает ту же историю. Просмотрите доступные учебные пакеты. Пакет initial содержит события от share1 до share6; followup содержит share7 и share8:
cat /home/labex/project/data/group-batches.json
Предоставленный инструмент публикации отправляет выбранные учебные данные на настоящий брокер, ждёт вызовов обратной связи о доставке и завершается с ошибкой при неполной доставке. Опубликуйте шесть начальных событий только после установления стабильного назначения:
/home/labex/project/tools/publish-shared-batch initial
/home/labex/project/tools/wait-caught-up orders-team 3 > /home/labex/project/output/team-initial-position.txt
cat /home/labex/project/output/team-initial-position.txt
Вспомогательный инструмент ожидания завершённого чтения только читает состояние и ждёт, пока у обоих разделов зафиксированное следующее смещение не станет равно 3, конечное смещение — 3, а отставание — 0. Каждый раздел получил три записи. Просмотрите, что прочитали два клиента:
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
Вместе их файлы должны содержать все шесть начальных идентификаторов. Отчёт о текущем назначении подтверждает стабильное владение; файл также может содержать более ранние результаты чтения до изменения назначения.
Теперь запустите отдельную группу orders-observer с ограничением числа сообщений и чтением с начала сохранённых данных:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 6 --timeout-ms 30000 --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/observer-initial.jsonl
cat /home/labex/project/output/observer-initial.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 3 > /home/labex/project/output/observer-initial-position.txt
cat /home/labex/project/output/observer-initial-position.txt
Она должна независимо получить все шесть записей. Она не отнимает разделы у orders-team, поскольку относится к другой группе. Обе группы могут одновременно дочитать данные до конца, сохраняя отдельные зафиксированные позиции.
Остановите одного участника и изучите оставшееся назначение
На этом шаге вы оставите reader-a работать, остановив только reader-b. kill отправляет сохранённому процессу сигнал завершения; брокер продолжает работать. wait позволяет оболочке завершить ожидание этого дочернего процесса. Процесс, получивший сигнал, может вернуть ненулевой код завершения, поэтому завершающее || true не даёт считать эту намеренную команду остановки ошибкой приложения:
kill "$consumer_b_pid"
wait "$consumer_b_pid" || true
/home/labex/project/tools/wait-assignment 1 > /home/labex/project/output/one-member.txt
cat /home/labex/project/output/one-member.txt
После стабилизации группы ожидается, что reader-a будет владеть обоими разделами. Инструмент ожидания проверяет настоящий отчёт перед следующей публикацией. При двух разделах добавление более двух участников оставило бы некоторых из них без раздела при стабильном назначении; в этой работе третий клиент не требуется.
Опубликуйте по одному дополнительному событию в каждый раздел и дождитесь, пока оставшийся участник дочитает данные:
/home/labex/project/tools/publish-shared-batch followup
/home/labex/project/tools/wait-caught-up orders-team 4 > /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/reader-a.jsonl
Теперь файл reader-a должен содержать и share7, и share8. reader-b остановлен, поэтому он не может обработать эти новые записи. У группы наблюдателя по-прежнему сохранена прежняя зафиксированная позиция; продолжите её чтение для двух новых событий:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 2 --timeout-ms 30000 --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/observer-followup.jsonl
cat /home/labex/project/output/observer-followup.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 4 > /home/labex/project/output/observer-followup-position.txt
cat /home/labex/project/output/observer-followup-position.txt
Ожидаются share7 и share8 и собственное следующее смещение каждой группы, равное 4 в обоих разделах. Параметр чтения с начала не отменяет допустимое сохранённое смещение 3 группы наблюдателя.
Перебалансировка может привести к повторной доставке, если выходящий участник обработал записи до фиксации позиции. Поэтому результаты чтения событий могут содержать повторы; при стабильном назначении у раздела всё равно один владелец в каждый момент времени. В этой работе проверяются фактические назначения и полное содержимое событий без заявления о сквозной гарантии доставки ровно один раз.
Остановите оставшийся клиент и сохраните подтверждения
На этом шаге вы завершите длительный эксперимент, остановив единственный оставшийся клиент. Вы уже проверили, что его конечные позиции достигли конца данных, поэтому необработанных записей нет. Используйте сохранённый PID именно этого клиента:
kill "$consumer_a_pid"
wait "$consumer_a_pid" || true
/home/labex/project/tools/wait-assignment 0 > /home/labex/project/output/no-members.txt
cat /home/labex/project/output/no-members.txt
Теперь у группы не должно быть активных участников. Сохранённые смещения и события брокера остаются; время жизни клиента и сохранённая позиция независимы. Клиенты наблюдателя с ограничением числа сообщений уже завершились автоматически.
Просмотрите сохранённые отчёты об участниках до и после перехода разделов:
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
Первый отчёт показывает двух владельцев с непересекающимися назначениями, а второй — одного владельца обоих разделов. Это фактические отчёты с идентификаторами участников, зависящими от запуска, а не фиксированные назначения для запоминания. Файлы читателей и наблюдателя сохраняют сведения о событиях для обеих групп. Вы остановили только два своих консольных клиента, сохранив топик и зафиксированные позиции для проверки.
Итоги
Вы увидели, как два потребителя делят фактические назначения разделов, отдельная группа независимо читает те же записи, а оставшийся участник получает оба раздела до поступления новой работы. Проверки состояния с ограниченным временем ожидания заменили догадки о задержках, а остановка конкретных процессов завершила клиенты, сохранив позиции и события. Далее вы замените подготовленный инструмент публикации производителем на Python, сериализацию и обработку доставки которого сможете изучить и изменить.
