Введение
Kafka хранит записи независимо от того, кто их прочитал. Группе потребителей также нужно запоминать свою позицию, чтобы последующий процесс мог продолжить чтение, а не начинать заново. Зафиксированное смещение — это следующая позиция, с которой группа должна читать в разделе, а не номер последней записи.
Вы прочитаете две из четырёх подготовленных записей, изучите зафиксированную позицию и отставание, продолжите чтение той же группой и сравните результат с новой группой. Затем вы предварительно просмотрите и выполните сброс, пока исходная группа неактивна, и снова прочитаете сохранённые записи. В топике один раздел, чтобы было проще разобраться в позициях.
В этой независимой работе предоставлены брокер, четыре записи и вспомогательная команда, которая опрашивает фактическое состояние координатора до появления ожидаемого смещения. Достаточно базовых навыков работы в терминале; писать код Python и настраивать сервисы не нужно. Вспомогательная команда не меняет смещения и не создаёт вымышленный отчёт. Используйте Terminal.
Прочитайте две записи и изучите зафиксированную следующую позицию
На этом шаге вы свяжете ограниченное чтение с фактической сохранённой позицией группы. Просмотрите четыре подготовленных значения:
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
Их смещения в этом новом топике с одним разделом — от 0 до 3. У новой группы orders-review нет сохранённой позиции, поэтому --from-beginning выбирает начало сохранённых данных для первого чтения.
Параметр --command-property передаёт значение конфигурации клиенту потребителя, а не меняет формат отображения событий. В этой работе консольный клиент автоматически фиксирует позицию. max.poll.records=1 ограничивает каждый опрос одной возвращаемой записью, чтобы остановка после двух показанных записей не продвинула позицию на весь более крупный возвращённый пакет. Интервал задаёт периодичность фиксации; завершающийся клиент также заканчивает обычную обработку позиции. Вы изучите фактический результат, не предполагая конкретной задержки:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/first-two.txt
cat /home/labex/project/output/first-two.txt
Ожидаются два события created, после чего клиент завершается. Запустите предоставленную вспомогательную команду опроса с ограниченным временем ожидания, указав имя группы и ожидаемое следующее смещение:
/home/labex/project/tools/wait-group-offset orders-review 2 > /home/labex/project/output/after-first.txt
cat /home/labex/project/output/after-first.txt
Команда повторно запрашивает описание настоящей группы, пока её зафиксированное смещение не станет равно 2, либо завершается с ошибкой по истечении срока ожидания. Она ничего не фиксирует и не сбрасывает. Найдите CURRENT-OFFSET 2, LOG-END-OFFSET 4 и LAG 2 для раздела 0. Сообщение об отсутствии активных участников группы нормально после завершения клиента с ограничением; сохранённая позиция по-прежнему существует.
Смещение 2 означает, что следующее чтение начнётся с третьей сохранённой записи. Конечное смещение 4 — это следующая позиция после четырёх существующих записей; отставание 4 − 2 = 2 измеряет число оставшихся записей в этом простом примере с неизменным состоянием. Первые две записи не удалились.
Продолжите чтение той же группой и сравните с новой
На этом шаге вы предположите, какие записи группа получит с сохранённой позиции. Снова запустите ту же группу, намеренно сохранив параметр чтения с начала:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/resumed-two.txt
cat /home/labex/project/output/resumed-two.txt
Ожидаются два события paid, а не предыдущие события created. --from-beginning задаёт начальную политику, когда допустимого зафиксированного смещения нет; он не отменяет допустимое сохранённое смещение 2 этой группы. Подтвердите завершение по фактическому состоянию координатора:
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-resume.txt
cat /home/labex/project/output/after-resume.txt
Ожидаются зафиксированное смещение 4, конечное смещение 4 и отставание 0. Теперь предположите результат для другой группы orders-independent. У неё нет сохранённой позиции, и она не зависит от orders-review, поэтому чтение с начала должно вернуть все четыре сохранённые записи:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-independent --from-beginning --max-messages 4 --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/new-group-all.txt
cat /home/labex/project/output/new-group-all.txt
/home/labex/project/tools/wait-group-offset orders-independent 4 > /home/labex/project/output/new-group-position.txt
cat /home/labex/project/output/new-group-position.txt
Ожидаются все четыре значения и собственное зафиксированное смещение новой группы, равное 4. Обе группы могут достичь конца, пройдя разную историю чтения. Позиция одной группы не сдвинула начальную позицию другой, и чтение ни одной из групп не удалило записи топика.
Далее вы намеренно измените сохранённую позицию исходной группы, а не будете считать параметр чтения с начала командой перемотки.
Просмотрите сброс неактивной группы и повторно прочитайте историю
На этом шаге вы сбросите только orders-review. Её потребитель с ограничением уже завершился, поэтому у группы нет активных участников. Перед сбросом остановите любого случайно оставленного работающим участника этой же группы; активный потребитель может продолжать менять позицию.
Параметр --reset-offsets выбирает операцию управления позицией, --topic ограничивает её область действия, а --to-earliest выбирает начало сохранённых данных. Сначала используйте --dry-run, чтобы просмотреть предлагаемое новое смещение без его применения:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --dry-run > /home/labex/project/output/reset-preview.txt
cat /home/labex/project/output/reset-preview.txt
Ожидается предлагаемое значение NEW-OFFSET, равное 0. Убедитесь, что предварительный просмотр не изменил текущую сохранённую позицию:
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-preview.txt
cat /home/labex/project/output/after-preview.txt
Теперь примените этот сброс в заданной области с помощью --execute:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --execute
/home/labex/project/tools/wait-group-offset orders-review 0 > /home/labex/project/output/after-reset.txt
cat /home/labex/project/output/after-reset.txt
Теперь фактическое зафиксированное следующее смещение должно быть равно 0, конечное смещение — 4, а отставание — 4. Записи остались на месте; изменилась только сохранённая позиция этой группы. Снова прочитайте все четыре записи, используя теперь допустимое зафиксированное начало без опоры на параметр начальной политики:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --max-messages 4 --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/replayed-all.txt
cat /home/labex/project/output/replayed-all.txt
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-replay.txt
cat /home/labex/project/output/after-replay.txt
Ожидаются исходные четыре значения и снова зафиксированное смещение 4. У orders-independent должна сохраниться собственная позиция завершённого чтения; сброс не вернул эту группу назад.
При повторном чтении существующие записи намеренно читаются ещё раз. Сбой между обработкой записи и фиксацией позиции также может привести к повторной доставке записи. Это упражнение с консолью демонстрирует позиции чтения, а не атомарную транзакцию записи в файл и фиксации смещения или сквозную гарантию обработки ровно один раз. В последующих работах с Python порядок обработки и фиксации будет задан явно.
Итоги
Вы изучили зафиксированные смещения следующего чтения и отставание, продолжили чтение с сохранённой позиции вопреки начальной политике и сравнили независимую группу. Предварительный просмотр и выполнение сброса неактивной группы позволили намеренно прочитать данные повторно, сохранив фактические записи топика неизменными. Далее вы понаблюдаете, как несколько участников одной группы распределяют разделы и восстанавливают назначение после выхода одного участника.
