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

На этом этапе вы разберёте поле request_api_version в заголовке запроса и вернёте код ошибки, если версия недопустима.

API Kafka

Каждый запрос Kafka — вызов API. Протокол Kafka определяет более 70 разных API, каждый со своим назначением. Например:

• Produce: записывает события в разделы.
• CreateTopics: создаёт новые топики.
• ApiVersions: возвращает поддерживаемые брокером версии API.

Запрос Kafka указывает вызываемый API в поле заголовка request_api_key.

Тело сообщения

Напомним формат запросов и ответов Kafka:

1. message_size.
2. Заголовок.
3. Тело.

До сих пор вы работали только с message_size и заголовком. Теперь начнёте работать с телом сообщения.

Каждый API определяет схемы тел своих запросов и ответов.

Например, тело запроса Produce API содержит такие поля:

• Имя топика для записи.
• Ключ раздела для записи.
• Данные события для записи.

Тело ответа Produce содержит коды результатов, указывающие, успешна ли каждая запись.

Версии API

Каждый API поддерживает несколько версий, позволяя схемам изменяться со временем. Версионирование работает так:

• Запрос указывает свою версию в поле заголовка request_api_version.
• Если версия поддерживается, ответ использует ту же версию. Например, на Produce Request (Version: 3) должен прийти Produce Response (Version: 3).
• История версий каждого API независима. Produce (Version: 10) не связан с Fetch (Version: 10).

API ApiVersions

API ApiVersions возвращает поддерживаемые брокером версии API. Например, он может сообщить о поддержке Produce версий 5-11, Fetch версий 0-3 и т. д.

На этом этапе вы начнёте добавлять поддержку ApiVersions версии 4. Пока достаточно поля error_code; остальные поля вы реализуете позже.

Поле error_code

Тело ответа ApiVersions начинается с 16-битного целого числа со знаком — error_code. Оно указывает, произошла ли ошибка при обработке запроса. Если ошибки нет, значение равно 0. Все возможные значения перечислены в таблице кодов ошибок.

На этом этапе нужно обработать только код 35 (UNSUPPORTED_VERSION). Эта ошибка возникает, когда клиент запрашивает неподдерживаемую брокером версию ApiVersions. Можно считать, что брокер поддерживает версии 0-4.

Тесты

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

$ ./your_program.sh

Затем подключится к брокеру на порту 9092 и отправит запрос ApiVersions с неподдерживаемой версией:

$ echo -n "000000230012674a4f74d28b00096b61666b612d636c69000a6b61666b612d636c6904302e3100" | xxd -r -p | nc localhost 9092 | hexdump -C

Запрос выглядит так:

00 00 00 23  // message_size:        35
00 12        // request_api_key:     18
67 4a        // request_api_version: 26442
4f 74 d2 8b  // correlation_id:      1333056139
...

Брокер должен вернуть ответ ApiVersions версии 4 с полем error_code, равным 35:

00 00 00 00  // message_size:   0 (any value works)
4f 74 d2 8b  // correlation_id: 1333056139
00 23        // error_code:     35

Для допустимой версии (0-4) брокер должен возвращать код ошибки 0, хотя на этом этапе это не проверяется.

Примечания

• API протокола Kafka отличаются от основных API Kafka. Основные API — более высокоуровневые интерфейсы Java и Scala, оборачивающие протокол Kafka.
• Можно считать, что тестирующая система отправляет только запросы ApiVersions. Проверять поле request_api_key в заголовке не нужно.
• На этом этапе тестирующая система проверяет только длину поля message_size — 4 байта, но не его значение. Правильные значения message_size вы реализуете позже.