Kafka tombstone и JSON null: когда событие удаляет запись

Consumer получает обновления профилей. У одного сообщения само value отсутствует, у другого value содержит JSON с полем email: null. Если обработчик проверяет только «пустое значение», оба случая могут превратиться в удаление профиля. Но это разные уровни данных.

В компактируемой теме Kafka запись с ключом и null value служит tombstone — отметкой удаления значения этого ключа. JSON-объект с отдельным null остаётся обычным непустым значением сообщения. Строка JSON null тоже представляет байты, а не отсутствие Kafka value.

Проверим транспорт до разбора JSON

В примере обработчик получает исходные key и value в байтах, как их отдаёт клиент до пользовательской десериализации. None означает отсутствующее value. Для обычного обновления контракт требует JSON-объект; запись с JSON null отклоняется.

import json

state = {}
def apply_record(key_bytes, value_bytes):
    if key_bytes is None:
        raise ValueError("Profile key required")
    key = key_bytes.decode("utf-8")
    if value_bytes is None:
        state.pop(key, None)
        return "deleted"
    value = json.loads(value_bytes)
    if not isinstance(value, dict):
        raise ValueError("Profile must be a JSON object")
    state[key] = value
    return "updated"

assert apply_record(b"A", b'{"email":"a@example.test"}') == "updated"
assert apply_record(b"A", b'{"email":null}') == "updated"
assert state == {"A": {"email": None}}
print("after null field", state)
assert apply_record(b"A", None) == "deleted"
assert state == {}
try:
    apply_record(b"B", b"null")
except ValueError:
    print("JSON null rejected")
else:
    raise AssertionError("JSON null was accepted")
print("after tombstone", state)

После обновления с email=null профиль A остаётся в состоянии с очищенным email. После настоящего tombstone профиль удаляется. Повтор удаления безопасен: pop с заданным значением по умолчанию не считает отсутствие ключа ошибкой.

Этот пример хранит состояние в памяти для проверки семантики. В рабочем consumer замените присваивание и удаление операциями целевого хранилища, затем фиксируйте offset после успешного результата. Не подтверждайте чтение только потому, что JSON удалось разобрать.

Не потеряйте различие в десериализаторе

Если слой клиента превращает и отсутствующее value, и JSON null в один None, бизнес-код уже не сможет понять исходный случай. Сохраните отдельный признак tombstone до разбора либо возвращайте разные типы результата. Для клиента confluent-kafka проверяйте message.value() до json.loads.

Пустые байты b'' тоже не являются tombstone: это значение нулевой длины, которое не подходит под наш JSON-контракт. Ошибка парсинга должна попасть в обработку некорректных событий, а не в удаление данных.

Компактация не заменяет обработчик удаления

Удаление старых версий выполняется фоновыми процессами Kafka, поэтому consumer может прочитать несколько обновлений одного ключа. Он должен применять их в порядке партиции и понимать tombstone сам. Срок хранения отметок удаления важен при восстановлении давно простаивавшего потребителя.

Проверьте четыре входа: обычный объект, объект с null, отсутствующее value и некорректный JSON. Для внешней базы добавьте сбой между записью и commit offset, затем повтор того же события. Так отдельно проверяются значение операции и устойчивость её выполнения.

В курсе Data Engineering Fast-Track вы разберёте темы, ключи и схемы Kafka, затем проведёте события через Spark в lakehouse. Откройте программу, чтобы научиться сохранять смысл обновления и удаления во всём конвейере.