Введение
Потребитель, который показывает событие, ещё не обязательно сохранил полезный результат для дальнейшей передачи. Это приложение декодирует каждое значение, сохранит его ключ и позицию Kafka в локальном файле JSONL и зафиксирует позицию только после успешной записи соответствующих данных и сброса буфера.
Вы перенесёте знакомую модель групп и смещений в Python: subscribe выбирает топик, poll запрашивает записи, а явная синхронная фиксация сохраняет следующее смещение для обработанного сообщения. Автоматическая фиксация и автоматическое сохранение смещений отключены, чтобы они не продвигали позицию независимо от вашего кода. Запись в файл и фиксация Kafka остаются отдельными операциями; это не атомарная транзакция между системами.
В этой независимой работе шесть начальных событий заказов, знакомых по работе с производителем, подготовлены в новом топике. Состояние виртуальной машины из предыдущих работ не требуется. Вам понадобятся функции, циклы, словари, JSON и запуск скриптов Python. Структура разбора аргументов и ограничения времени предоставлена. Используйте Terminal и установленную среду bigdata-python.
Подпишитесь, получите и декодируйте одну настоящую запись
На этом шаге вы просмотрите одну полученную запись без фиксации позиции предварительного просмотра. Сначала прочитайте подготовленные значения событий:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Consumer настраивается с идентификатором группы и начальной политикой earliest. subscribe регистрирует нужный топик; poll возвращает сообщение, сообщение с ошибкой или None, когда запись недоступна, в том числе во время назначения разделов. Цикл продолжает опрос в пределах общего времени ожидания.
Предоставленная структура управления ограничивает чтение по времени. time.monotonic() измеряет прошедшее время независимо от календарных часов. continue начинает следующий опрос, если сообщение ещё не готово; break выходит из цикла после вывода записи. Ветвь else цикла while выполняется, только если цикл завершился без этого break, что здесь означает отсутствие записи до истечения срока ожидания.
finally закрывает потребителя как при успешном чтении, так и при возникновении ошибки. Рассматривайте эту структуру как ожидание и освобождение ресурсов вокруг вызовов Kafka; проектировать механизм повторных попыток не нужно.
В этом предварительном просмотре используется отдельная группа file-preview с отключёнными обеими автоматическими настройками. Декодированное значение — словарь Python; обёртка также сохраняет топик, раздел, смещение и декодированный ключ. Вызова commit нет:
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
Ожидается одна обёртка JSON, значение которой содержит одно из событий от py1 до py6, а ключ совпадает с order_id этого события. При трёх разделах первое прочитанное событие не фиксировано. Раздел и смещение определяют фактическую запись на брокере.
Закрытие этого предварительного просмотра не фиксирует позицию, поскольку автоматическая фиксация отключена, а скрипт не делал явной фиксации. Следующее приложение использует собственную группу для передачи данных в файл.
Записывайте каждую обёртку и сбрасывайте буфер до фиксации сообщения
На этом шаге вы создадите потребителя с ограничением чтения для записи в файл. Внимательно изучите порядок обработки: декодирование и проверка объекта JSON и ключа, запись одной обёртки JSONL, сброс буфера файла, синхронная фиксация этого сообщения и только затем учёт его как завершённого. Если запись, декодирование или фиксация завершаются ошибкой, приложение прекращает работу, а не пропускает ошибку с последующим ложным сообщением об успешном пакете.
Оператор with закрывает локальный файл при выходе из его блока. Path указывает имя локального файла. Выходной файл открывается в режиме добавления, чтобы перезапуск не обрезал молча предыдущие подтверждения доставки. При первом чистом запуске файл ещё отсутствует. max-messages — целевое число сообщений для этого запуска; общий срок ожидания приводит к ненулевому коду завершения при неполном чтении. JSONL хранит один полный объект JSON в каждой строке без окружающего массива. flush() передаёт буферизированные данные файла из Python операционной системе перед запросом фиксации. Это не гарантирует сохранность при сбое питания или диска и не превращает запись файла и фиксацию Kafka в одну атомарную транзакцию. В этой работе исследуются прерывание процесса и сохранённая позиция в работающей виртуальной машине.
commit(message=..., asynchronous=False) фиксирует следующее смещение обработанного сообщения для его раздела. Возвращаемые ошибки для отдельных разделов проверяются. enable.auto.offset.store=false предотвращает постороннее автоматическое сохранение позиции, а явный аргумент message точно указывает обрабатываемую позицию для фиксации.
Сохраните полную предоставленную структуру разбора аргументов и цикла с этой последовательностью обработки:
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Ожидаются Saved and committed records: 6 и шесть обёрток JSONL. Каждая обёртка должна сохранять полное значение, соответствующий ключ и фактическую позицию записи. Обратный вызов или консольный журнал не заменяют сохранённый файл; на следующем шаге вы сравните содержимое файла с фактической зафиксированной позицией.
Свяжите файл с зафиксированными следующими смещениями разделов
На этом шаге вы изучите позицию на координаторе с помощью уже знакомого инструмента командной строки:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
Для каждого раздела с доставленными записями CURRENT-OFFSET должен достичь LOG-END-OFFSET этого раздела, а отставание должно быть нулевым. Конкретное назначение разделов и распределение записей в этой работе не фиксированы. У раздела без записей пока может не быть зафиксированной позиции; в нём нет обработанного сообщения, которое можно зафиксировать.
Сравните смещение сохранённой обёртки с зафиксированным следующим смещением её раздела. Если последняя обработанная запись раздела имеет смещение n, следующая зафиксированная позиция — n+1. Эта следующая позиция не является общим счётчиком по всем разделам.
Потребитель закрылся, поэтому сообщение об отсутствии активных участников нормально. Сохранённый файл и позиция группы остаются. У вас есть отдельные подтверждения результата обработки и состояния координатора без утверждения, что они записаны в одной транзакции.
Перезапустите ту же группу и убедитесь в отсутствии новых записей
На этом шаге вы проверите сохранённую позицию вместо предположения, что earliest всегда возвращает чтение назад. Шесть записей по-прежнему хранятся, но у file-writer есть допустимые зафиксированные конечные позиции. Предположите результат ещё одного запуска с той же группой: новых записей не должно быть, хотя auto.offset.reset задан как earliest.
Используйте отдельный путь вывода, чтобы сохранить исходный файл из шести записей. Запросите одну запись с коротким общим сроком ожидания. Поскольку запрошенного числа достичь нельзя, приложение намеренно сообщает о неполном пакете ненулевым кодом завершения. Следующая команда должна завершиться с ошибкой.
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
Сразу сохраните код завершения:
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
Ожидаются ненулевой код завершения и Only received 0 of 1 requested records before the deadline. Новый файл должен быть пустым; исходный received-events.jsonl по-прежнему содержит шесть обёрток. Допустимые зафиксированные смещения той же группы не были заменены начальной политикой earliest.
Чтобы намеренно прочитать сохранённую историю ещё раз, используйте уже изученный подход с отдельной группой или сбросом неактивной группы. Если процесс останавливается после записи, но до фиксации, ситуация иная: та же группа может получить запись повторно. Следующая работа наглядно покажет этот промежуток, в котором возможен сбой.
Итоги
Вы подписались на топик и получали записи с помощью Python, декодировали JSON/UTF-8 и сохраняли ключи и позиции Kafka в JSONL. Успешный сброс буфера предшествовал явной синхронной фиксации, а фактическое состояние координатора соответствовало заполненному файлу. Перезапуск той же группы не обнаружил новых записей, при этом сохранённая история топика не удалялась. Далее ошибка декодирования и прерывание между записью и фиксацией покажут, почему записи могут доставляться повторно.
