Введение
Успешный потребитель записал файл, а затем зафиксировал позицию чтения. Это две отдельные операции. В этой работе вы увидите два случая ошибки: некорректная запись JSON должна остановить обработку до продвижения позиции за неё, а процесс, завершившийся после записи, но до фиксации, может снова получить ту же запись.
Сначала вы выполните обычную доставку, затем изучите точную позицию некорректной записи и, наконец, с помощью предоставленного учебного параметра остановите процесс между сбросом буфера и фиксацией. Перезапуск той же группы покажет повторную доставку в файле по совпадению идентификатора события и позиции Kafka. Вы понаблюдаете за этим поведением без реализации устранения дубликатов или механизма повторных попыток.
В этой независимой среде подготовлены три топика с одним разделом и полный потребитель, знакомый по предыдущей работе. Дополнительный параметр управляемого выхода уже реализован. Вам понадобятся изученные ранее понятия JSON, файлов и ручной фиксации в Python; перенастройка брокера и новая архитектура восстановления не требуются. Используйте Terminal.
Проверьте успешный путь обработки
На этом шаге вы убедитесь, что приложение работает с корректными входными данными, прежде чем разбирать ошибку. У каждого топика один раздел, поэтому позиции записей легко проследить. orders-normal содержит две корректные записи. Потребитель использует ту же явную последовательность записи, сброса буфера, синхронной фиксации сообщения и закрытия, что и раньше.
Просмотрите предоставленное приложение, включая параметр управляемого выхода, ничего не меняя:
cd /home/labex/project
cat /home/labex/project/consumer.py
По умолчанию этот параметр выключен. Выполните обычную обработку двух записей с отдельной группой и файлом:
bigdata-python /home/labex/project/consumer.py --topic orders-normal --group normal-reader --output /home/labex/project/output/normal-delivery.jsonl --max-messages 2 --timeout-seconds 20 > /home/labex/project/output/normal.log 2>&1
cat /home/labex/project/output/normal.log
cat /home/labex/project/output/normal-delivery.jsonl
Ожидаются normal1 и normal2 со смещениями 0 и 1, а после обеих записей — следующее зафиксированное смещение 2. max-messages — число сообщений, запрошенное в этом запуске; файл в режиме добавления может сохранять результаты нескольких запусков на последующих шагах.
Этот успешный исходный результат позволяет отделить проблему соединения всего приложения от ошибки отдельной записи. Для намеренных ошибок вы используете разные топики, группы и файлы, чтобы подтверждения оставались понятными.
Остановитесь на некорректном JSON, не фиксируя позицию за ним
На этом шаге вы изучите корректную начальную запись, некорректную запись JSON и последующую корректную запись. Прочитайте локальные данные:
cat /home/labex/project/data/bad-events.txt
В разделе 0 топика orders-bad корректная начальная запись находится по смещению 0, некорректный текст — по смещению 1, а последующее корректное событие — по смещению 2. Политика — немедленная остановка: сообщить позицию ошибки и завершиться. Не перехватывайте ошибку с продолжением фиксации последующих записей этого раздела.
Запросите доставку трёх записей. Следующая команда должна завершиться с ошибкой.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-first.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first.log
cat /home/labex/project/output/bad-prefix.jsonl
Ожидаются ненулевой код завершения и Decode failure с указанием topic=orders-bad partition=0 offset=1. Файл должен содержать только bad-prefix со смещением 0. Его успешная запись была зафиксирована, поэтому следующее смещение группы равно 1; некорректная и последующая записи остаются необработанными.
Перезапустите ту же группу с добавлением в тот же файл. Следующая команда должна завершиться с ошибкой.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-retry.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry.log
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group bad-reader --describe
То же смещение 1 должно снова вызвать ошибку. Корректная начальная запись не должна записаться повторно, поскольку следующая после неё позиция 1 уже зафиксирована. Последующее корректное событие не должно появиться в файле. Так политика становится наглядной: оператор должен выбрать способ работы с некорректными данными, а это минимальное приложение не пропускает их молча и не фиксирует позицию за ними.
Завершите процесс после успешной записи, но до её фиксации
На этом шаге вы создадите другой промежуток возможного сбоя с подготовленным топиком orders-replay. Он содержит только repeat1 со смещением 0. Учебный параметр --exit-before-commit завершает процесс с кодом 73 сразу после сброса буфера файла и до явной фиксации. Он выводит фактическую позицию и по-прежнему закрывает потребителя с отключённой автоматической фиксацией.
Используйте новую группу и отдельный файл. Следующая команда должна завершиться с ошибкой.
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 --exit-before-commit > /home/labex/project/output/before-commit-exit.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit.log
cat /home/labex/project/output/redelivery.jsonl
Ожидаются код 73 и одна сохранённая обёртка repeat1 в разделе 0 со смещением 0. Журнал содержит CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. Это предоставленная демонстрация сбоя, а не изменение, которое вам нужно реализовать.
Теперь запись есть в файле, но у группы нет зафиксированной позиции для неё. Поэтому при следующем запуске той же группы применяются изученные ранее правила начальной политики. Именно это расхождение вы увидите на следующем шаге.
Перезапустите потребителя и определите повторно доставленную запись Kafka
На этом шаге вы перезапустите replay-reader без учебного параметра. Отсутствие зафиксированной позиции и политика earliest позволяют ему снова прочитать сохранённую запись. Сохраните режим добавления и тот же выходной файл, чтобы обе доставки оставались видны:
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 > /home/labex/project/output/after-restart.log 2>&1
cat /home/labex/project/output/after-restart.log
cat /home/labex/project/output/redelivery.jsonl
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay-reader --describe
Ожидаются две обёртки с одинаковыми event_id, ключом, значением, топиком, разделом и смещением. На брокере по-прежнему только одна исходная запись. Второй запуск успешно фиксирует следующее смещение 1, а файл сохраняет подтверждения обеих доставок.
Используйте следующие сведения, чтобы различать типичные ситуации:
| Наблюдение | Первые полезные сведения |
|---|---|
| Нет новых событий | Корректный топик и позиция группы в конце; опрос с ограничением времени может не вернуть записей |
| Неверное имя топика | Сравните указанное имя со списком или описанием топиков; этот клиент отключает случайное создание топиков |
| Нет допустимой зафиксированной позиции | Изучите позицию группы и заданную начальную политику earliest; она может прочитать сохранённую историю |
| Ошибка декодирования | Изучите указанные топик, раздел, смещение и данные значения; группа останавливается до продвижения за некорректную позицию |
Запись в файл и фиксация позиции Kafka не являются одной транзакцией. Прерывание между ними может вызвать повторную доставку, что и показывает фактическая позиция этой записи. В работе сохраняются обе доставки без реализации устранения дубликатов, очереди недоставленных сообщений или транзакционной архитектуры. В заключительном задании вы примените успешный путь с явным порядком повторной попытки через чистый запуск.
Итоги
Вы выполнили обычную доставку, остановились на некорректном JSON без фиксации позиции за ним и увидели повторную доставку записи после перезапуска, когда её буфер уже был сброшен, а смещение ещё не зафиксировано. Совпадение содержимого события и позиции Kafka объяснило повторные записи в файле без предположения о втором событии на брокере. Заключительное задание объединяет успешную публикацию, ограниченное чтение, вывод в файл и запись перед фиксацией на новых данных.
