Формат сообщений в системном топике Kafka

Содержание раздела

1. Формат ключа сообщения
2. Формат тела сообщения

Система записывает в топик Kafka сообщения по следующим событиям:

• создание, изменение или удаление логической сущности (кроме внешних таблиц);
• открытие дельты;
• закрытие дельты;
• откат дельты;
• завершение операции записи;
• отмена операции записи.

Сообщения записываются в JSON-формате в топик, имя которого задано с помощью параметра конфигурации KAFKA_STATUS_EVENT_TOPIC. Каждое сообщение состоит из ключа и тела, их форматы описаны ниже.

Запись событий о завершении и отмене операций записи в топике можно отключить, установив для параметра конфигурации KAFKA_STATUS_EVENT_WRITE_OPERATIONS_ENABLED значение false.

Рекомендуется настроить автоматическую очистку системного топика в конфигурации брокера Kafka.

Формат ключа сообщения

Ключ сообщения в системном топике Kafka имеет следующий формат:

{
  "datamart": "<logical_db>",
  "datetime": "<event_datetime>",
  "event": "<event_code>",
  "eventLogId": "<event_UUID>"
}

Где:

• logical_db — имя логической базы данных, в которой произошло событие;
• event_datetime — дата и время события;
• event_code — код события. Возможные значения:
  • DATAMART_SCHEMA_CHANGED — изменение логической схемы (кроме создания и удаления внешних таблиц),
  • DELTA_OPEN — открытие дельты,
  • DELTA_CLOSE — закрытие дельты,
  • DELTA_CANCEL — откат дельты,
  • WRITE_OK — завершение операции записи,
  • WRITE_CANCEL — отмена операции записи.
• event_UUID — уникальный идентификатор события.

Пример ключа в сообщении о закрытии дельты:

{
  "datamart": "marketing", 
  "datetime": "2022-02-28 08:02:40",
  "event": "DELTA_CLOSE",
  "eventLogId": "6b80c392-3198-492f-a631-6a684e521b79"
}

Формат тела сообщения

Тело сообщения в системном топике Kafka содержит параметры события и имеет формат, описанный в таблице ниже.

Событие	Формат тела сообщения
Изменение логической схемы	{"changeDateTime": "<change_datetime>", "datamart": "<logical_db>"}
Открытие дельты	{"deltaNum": <delta_number>}
Закрытие дельты	{"deltaNum": <delta_number>, "deltaDate": "<delta_commit_datetime>"}
Откат дельты	{"deltaNum": <delta_number>}
Завершение операции записи	{"entity": <entity_name>, "cn": <sys_cn>, "ts": '<ts_datetime>', "rowsAffected": null}
Отмена операции записи	{"entity": <entity_name>, "cn": <sys_cn>}

Где:

• change_datetime — дата и время изменения в формате YYYY-MM-DD hh:mm:ss;
• logical_db — имя логической базы данных, к которой относится изменение;
• delta_number — номер дельты;
• delta_commit_datetime — дата и время закрытия дельты в формате YYYY-MM-DD hh:mm:ss.SSSSSS;
• entity_name — имя логической сущности, в которой выполнена или отменена операция записи;
• sys_cn — номер операции записи. Для прокси-таблиц и standalone-таблицы указывается значение null;
• ts_datetime — дата и время выполнения или отмены операции записи в формате YYYY-MM-DD hh:mm:ss.SSSSSS. Для прокси-таблиц и standalone-таблицы указывается значение null;
• rowsAffected — поле, зарезервированное для будущего использования.

Пример тела в сообщении об изменении логической схемы:

{
  "changeDateTime": "2022-02-24 15:25:47",
  "datamart": "marketing"
}

Пример тела в сообщении о закрытии дельты:

{
  "deltaNum": 8,
  "deltaDate": "2022-02-28 08:02:40.420923"
}