Собственная Kafka

На этом этапе вы реализуете ответ Produce для существующих топиков и разделов.

Ответ API Produce для существующего топика

Получив запрос Produce, брокер Kafka должен проверить существование топика и раздела. Если хотя бы один из них не существует, брокер возвращает соответствующий ответ с кодом ошибки.

Проверка выполняется в следующем порядке:

1. Проверка топика: прочитать файл журнала топика __cluster_metadata и проверить, существует ли топик.
2. Проверка раздела: если топик существует, проверить наличие раздела в этом топике.

Проверка топика

Чтобы проверить существование топика, брокер читает файл журнала топика __cluster_metadata по пути /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000.log. В нём брокер находит метаданные топика — запись record внутри RecordBatch с полезной нагрузкой типа TOPIC_RECORD. Если существует TOPIC_RECORD с заданными именем и идентификатором топика, топик существует.

Проверка раздела

Чтобы проверить существование раздела, брокер читает тот же файл журнала __cluster_metadata и находит метаданные раздела — запись record внутри RecordBatch с полезной нагрузкой типа PARTITION_RECORD.

Раздел существует, если найдена запись PARTITION_RECORD, в которой совпадают:

• Запрошенный идентификатор раздела.
• UUID топика, которому принадлежит раздел.
• UUID каталога журнала, где хранится раздел.

Мы также создали интерактивный обозреватель файла журнала топика __cluster_metadata:

• 🔎 Файл журнала метаданных кластера

Тесты

Тестирующая система запустит вашу программу следующим образом:

$ ./your_program.sh /tmp/server.properties

Затем подключится к серверу на порту 9092 и отправит запрос Produce (v11) для существующих топика и раздела.

Она проверит, что:

• Первые 4 байта ответа («длина сообщения») корректны.
• Идентификатор корреляции в заголовке ответа совпадает с идентификатором в заголовке запроса.
• Поле throttle_time_ms ответа равно 0.
• Поле topics содержит 1 элемент, в котором:
  • Поле name совпадает с именем топика в запросе.
  • Поле partitions содержит 1 элемент, в котором:
    • Поле error_code равно 0 (NO_ERROR).
    • Поле index совпадает с разделом в запросе.
    • Поле base_offset равно 0, обозначая первую запись в разделе.
    • Поле log_append_time_ms равно -1, обозначая последнюю временную метку.
    • Поле log_start_offset равно 0.

Примечания

• Официальная документация API Produce находится здесь. Прокрутите до раздела «(Version: 11)».
• Официальная документация Kafka не описывает структуру записей топика __cluster_metadata, но определения можно найти в исходном коде Kafka.