org-little / kafka-object-creator-java
Форк
0
CI/CD
Реестр контейнеров и пакетов
    Андрей
    3.3.1
    2 месяца назад История
    README.md

    kafka-object-creator-java

    Release Notes

    1.3.1

    • [FEATURE] Добавлена поддержка шифрования паролей инструментом little-kafka (см. Шифрование паролей в JAAS).
    • [FEATURE] Изменен формат вывода режима diff. Теперь печать реализуется развернутыми двухколоночными JSON-структурами (см. Режим верификации).
    • [FEATURE] Улучшен алгоритм сравнивания.
    • [FEATURE] Некоторые параметры аутентификации и авторизации разрешается передавать из командной оболочки (см. Другие опции, передаваемые из командной строки).
    • [MINOR] Версия формата YAML повышена до 1.0.5. В модель Yaml добавлена поддержка списка clusters (на будущее). В текущей реализации вставка списка не запрещается, но утилитой он игнорируется.
    • [MINOR FIX] Убрана лишняя печать сообщений в лог.

    1.2.1

    • [FIX] Небольшие дополнения в скрипт запуска start-koc.sh.

    1.2.0

    • [FEATURE] Добавлена функция выгрузки объектов с сервера в YAML-формате (см. Режим опроса сервера).

    1.1.4

    • [FEATURE] Добавлено логирование в файлы (см. Логирование событий в файлы).
    • [FEATURE] Добавлен параметр -Dbootstrap-server и переменная окружения KOC__BOOTSTRAP_SERVER для возможности передачи параметра подключения с командной строки.
    • [FIX] Исправления в скрипте start-koc.sh: учет особенностей запуска в Cygwin и на *nix системах.

    1.1.3

    • [FIX] Синхронизировано выполнение команд по созданию / удалению топиков (ранее всё выполнялось асинхронно), из-за задержек в выполнении заданий на медленных системах.

    1.1.2

    • [FEATURE] Добавлена поддержка для ACL-ресурсов cluster и delegation-token. (см. Формат YAML). Версия формата YAML повышена до 1.0.4.
    • [FEATURE] Добавлены псевдонимы для ключевых слов transactional-id и delegation-token из-за частых опечаток. Соответственно допустимы варианты transactional_id и delegation_token.
    • [FEATURE] Добавлена поддержка операции idempotentwrite.

    1.1.1

    • [FEATURE] Добавлена экспериментальная возможность изменять некоторые параметры топика при его создании или модификации (см. Формат YAML). Версия формата YAML повышена до 1.0.3.
    • [MINOR FIX] Исправлена ошибка, при которой не изменялся max.message.bytes для вновь создаваемого топика.

    1.1.0

    • [UPDATE] Обновление клиентской библиотеки Kafka до версии 3.9.0 и обновление прочих зависимых библиотек.
    • [FEATURE] Реализован Multi-YAML режим чтения входящих файлов (см. Правила обработки YAML файлов).
    • [FIX] Исправлено некорректное поведение expectations, при котором происходит накапливание значений от уже обработанных YAML-описаний. Теперь каждый expectations привязан к ближайшей к нему структуре (см. Правила обработки YAML файлов).
    • [MINOR FIX] Сообщение о создании/удалении существующего/несуществующего объекта понижено до уровня DEBUG, чтобы не захламлять вывод на консоль. Вместо этого добавлены новые метрики в итоговых статистиках (см. Итоговые статистики).
    • [MINOR FIX] Убран лишний вывод WARN-сообщения в моменты, когда передаваемый max.message.bytes равен фактическому.
    • [DOC] Добавлен раздел Известные ограничения утилиты и рекомендации.

    1.0.5

    • [FEATURE] Добавлен параметр конфигурации koc.kafka.connection.timeout.s, который определяет ожидание утилитой ответа брокера во время инициализации подключения, в секундах. Перед началом своей работы утилита делает пробный запрос на один из узлов Kafka для проверки того, что кластер активен. Чтобы не подвеситься, утилита ждет не более koc.kafka.connection.timeout.s секунд. Если в течение этого промежутка не последовало ответа, утилита завершает работу. Значение по умолчанию — 10 секунд.
    • [FEATURE] Добавлена возможность валидировать YAML конфигурацию перед передачей её на исполнение (опция -V или --validate-before-execute).
    • [FEATURE] Формат YAML дополнен новым списком expectations (версия формата повышена до 1.0.2. См. описание в секции Формат YAML).
    • [FEATURE] В формат объявления топика добавлен флаг recreate, чтобы уведомить утилиту пересоздать топик, если он существует. Флаг работает только с заданиями на создание топиков.
    • [FEATURE] Добавлена проверка на существование/несуществование объекта перед отправкой задания на создание/удаление.
    • [FEATURE] Добавлена дополнительная операция при создании топика: если при создании топика он существует, но при этом в YAML файле явно указан max.message.bytes и он превышает актуальное значение на сервере, утилита сделает попытку увеличить max.message.bytes на сервере. При этом утилита не позволит сделать уменьшение этого параметра (единственный вариант это пересоздать топик).
    • [FIX] В режиме верификации утилита запрашивает с сервера актуальный максимальный размер сообщения топика и подставляет его в определение объекта, если значение там не определено пользователем, что устраняет проблему сравнивания с -1.
    • [MINOR FIX] Утилита по умолчанию пишет свои сообщения в STDOUT, а не STDERR, как было до этого.
    • [MINOR FIX] Изменен порядок исполнения заданий: порядок исполнения заданий на создание и удаление топиков поменян местами: сначала выполняются задания на удаление топиков, а потом на создание.

    1.0.4

    • Исправлены некоторые небольшие ошибки.

    1.0.3

    • Реализован режим запуска верификации.
    • Исправлена сборка релизного архива.
    • Исправлены некоторые ошибки.

    1.0.2

    • Реализован базовый функционал:
      • Чтение формата CSV для плавного перехода с Bash версии утилиты.
      • Реализован формат Yaml версии 1.0.1 для описания создаваемых объектов (сам формат описан ниже на данной странице).
      • Формирование запросов к кластеру для создания объектов через клиентскую библиотеку kafka-clients-*.jar на основе передаваемых файлов описаний.

    Описание

    Утилита предназначена для автоматизированного создания Kafka объектов по спискам из файлов описаний. Утилита использует библиотеку kafka-clients-*.jar для создания/удаления следующих объектов:

    • топики;
    • ACL правила для:
      • топиков,
      • групп приемников,
      • транзакционных идентификаторов.

    Утилита принимает на вход конфигурационный файл (один или несколько) с описанием создаваемых объектов, затем парсит его и отправляет на его основе запросы на брокер. Для описания объектов используется два формата:

    • таблица CSV (устаревший формат);
    • структура YAML.

    Описание форматов приводятся ниже.

    Список зависимостей

    • kafka-clients — клиентская библиотека Apache Kafka.
    • csv-reader — библиотека чтения формата CSV. Поставляется вместе с утилитой.
    • slf4j + log4j — библиотека логирования.
    • commons-io + commons-lang3 + commons-cli — для некоторых утилитарных функций.
    • snakeyaml — библиотека чтения формата YAML.

    Все зависимости идут в поставке с архивом собранной утилиты.

    Работа с утилитой

    Установка

    Примечание:
    Сборка версии 1.0.2 имела ошибки в сборочном сценарии, поэтому утилиту из этой сборки нельзя запустить без особых инструкций. Используйте только версии старше 1.0.2.

    Для установки утилиты достаточно распаковать ZIP архив сборки в некоторую директорию. В архиве идут следующие каталоги и файлы:

    • libs/ — каталог с используемыми зависимостями.
    • etc/ — каталог с примерами конфигураций.
      • example.yaml — пример описания перечня объектов в YAML-формате.
      • config.properties — пример конфигурационного файла утилиты.
    • kafka-object-creator-java-<ВЕРСИЯ>.jar — jar-архив со скомпилированным кодом утилиты.
    • start-koc.sh — вспомогательный Bash-скрипт, через который можно запускать утилиту из командной строки Bash (не ниже версии 4).
    • README.md — Markdown-файл с документацией.

    Убедитесь, что запускающее окружение видит JRE (не ниже 1.8) и команду java. Если для запуска вы используете скрипт start-koc.sh, то по умолчанию он пытается работать относительно диспозиции, в которой находится файл сценария, и пытается найти kafka-object-creator-java-<ВЕРСИЯ>.jar рядом с собой, а конфигурацию утилиты по пути ./etc/config.properties. Некоторые диспозиции в этом скрипте можно менять через переменные окружения. На его основе вы можете реализовать свой сценарий запуска, либо запускать утилиту java-утилиту непосредственно.

    Вызов

    Утилита поставляется в формате self-executable-jar-archive, т.е. утилита запускается на основе манифеста. Код утилиты написан на Java 1.8.

    Общий вид команды для запуска утилиты:

    [KOC__REPLICATION_FACTOR=<число>] [KOC__DIFF_MODE=1] [KOC__INPUT=file1,file2,...] \
[KOC__VALIDATE=1] \
[KOC__GET_YAML=1] \
[KOC__BOOTSTRAP_SERVERS=<хост1>:<порт1>,<хост2>:<порт2>] \
java ... \
    -Dkoc.config="/path/to/config.properties" \
    [-Dreplication.factor=<число>] \
    [-Dbootstrap.servers="<server1>:<port1>,<server2>:<port2>"] \
    [-Dclient.id="koc-utility"] \
    [-Dsasl.mechanism="<sasl-mechanism-id>"] \
    [-Dsasl.jaas.config="<sasl-jaas-config-line>"] \
    [-Dsecurity.protocol="<security-protocol-id>"] \
  -jar kafka-object-creator-*.jar \
    [-h | --help] \
    [-d | --diff] \
    [-V | --validate-before-execute] \
    [--get-yaml [--no-print-replication-factor] [--dir /path/to/dir]] \
    [input.{yml|yaml|csv}] ...

    Утилита конфигурируется с помощью файла config.properties, в котором задаются все параметры конфигурации, в том числе параметры подключения клиента к брокеру. Если конфигурация не передается явно, утилита пытается найти файл с именем config.properties в диспозиции пакета org.little.koc. Если файл не удается найти, утилита завершается с кодом ошибки -1. Также утилита возвращает ненулевые коды в случае различных ошибок, сопровождаемые записью в логах.

    Параметр подключения клиентом к кластеру может быть передан одним из следующих способов (по приоритету проверки, от высокого к низкому):

    • В файле конфигурации, переданного параметром -Dkoc.config. Имя параметра — bootstrap.servers.
    • В параметре командной строки -Dbootstrap.servers.
    • В переменной окружения KOC__BOOTSTRAP_SERVERS.

    Формат строки во всех способах одинаковый, описанный в документации Kafka.

    В качестве аргументов передаются файлы с описанием объектов Kafka. Описания объектов должны передаваться во всех режимах кроме режима опроса сервера (опция --get-yaml). За один раз необходимо передать по меньшей мере один файл для нормальной работы утилиты.

    Файлы описания объектов могут быть переданы несколькими путями, в порядке проверки:

    • через конфигурационый файл (параметр csv.files);
    • через параметры командной оболочки (... [input.{csv|yml|yaml}] ...);
    • переменной окружения KOC__INPUT=file1.csv,file2,csv,file3.yml;
    • системной properties-переменной -Dkoc.file.list=file1.csv,file2.csv,file3.yml.

    Примечание:
    Утилита прекращает поиск списка, если на одном из этапов проверки он обнаруживается не пустым. По этой причине вы можете поставлять списки сразу несколькими способами, если вы используете макроподстановки в сценариях — срабатывать всегда будет только один.

    Если список не был получен ни одним из вышеописанных способов, утилита завершается с кодом ошибки -1.

    Примечание:
    При передаче списка последними двумя способами разделителем файлов служит символ запятой (,).

    Другие опции, передаваемые из командной строки

    С версии 1.3.0 допускается передача командной строкой нижеперечисленных опций, которые не имеют значений по умолчанию (сделано для удобства тестирования). Передача опций таким образом всегда ниже по приоритету, чем передача через конфигурационный файл. В реальных условиях отдавайте предпочтение конфигурационному файлу, так как доступ к нему можно ограничить правами доступа со стороны системы, что способствует секретности используемой схемы подключения.

    • -Dclient.id="<идентификатор-клиента>" — передает идентификатор, с которым клиент подключается к кластеру.
    • -Dsasl.mechanism="<идентификатор-механизма-авторизации>" — передает используемый механизм авторизации при подключении к кластеру. Используется только при SASL авторизации.
    • -Dsecurity.protocol="<идентификатор-безопасности>" — протокол безопасности, используемый клиентом для аутентификации.
    • -Dsasl.jaas.config="<JAAS-конфигурация>" — передает параметры авторизации в формате JAAS.

    Шифрование паролей в JAAS

    Пароль, передаваемый в SASL авторизации через JAAS, можно передавать в зашифрованном виде благодаря добавленному в версию 1.0.3 шифратору little-kafka. Чтобы зашифровать пароль, необходимо обратиться к библиотеке little-kafka, которая идет в поставке kafka-object-creator. Для простоты обращения к этой библиотеке, был подготовлен Bash-сценарий encryptPassword.sh, который имеет следующий синтаксис:

    ./encryptPassword.sh "<password>"

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

./encryptPassword.sh "<password_1>" "<password_2>" "<password_3>" ...

    В аргументе сценария вы передаёте пароль в открытом виде. Сценарий делает обращение к шифратору и возвращает шифрованное представление пароля, которое нужно скопировать. Например, если бы пароль в открытом виде был password, то вызов был бы таким:

    $ ./encryptPassword.sh "password"
original: password
MDS5
mask identifier: {mds5}R8086l3x+AZjymrbe1/eBA==

    Далее, в JAAS конфигурации для подключения к кластеру вы должны использовать модуль авторизации org.little.broker.auth.PlainLoginModule вместо модуля org.apache.kafka.common.security.plain.PlainLoginModule, которому следует передать полученный шифр так:

    sasl.jaas.config=org.little.broker.auth.PlainLoginModule required \
    username="admin" \
    password="{mds5}R8086l3x+AZjymrbe1/eBA==";

    Примечание:
    Данный приём работает только для схемы авторизации SASL_PLAINTEXT с механизмом PLAIN. При этом можно по-прежнему передавать пароль открыто с этим модулем, так как он опирается на маркер {mds5}, говорящий о том, что перед ним шифртекст.

    Режим верификации

    По умолчанию утилита запускается в режиме создания/удаления объектов. В утилиту внедрен второй режим работы (с версии 1.0.3), называемый режимом верификации. Утилита переключается в него, если:

    • переменная окружения KOC__DIFF_MODE, передаваемая утилите, определена любым значением;
    • если в вызове передается ключ -d или --diff.

    При этом прочий ввод для утилиты остается прежним. В этом режиме утилита запрашивает у Kafka кластера существующие объекты и производит сравнение с объектами, передаваемыми входными файлами описания объектов. Утилита покажет существование разницы только в случае, если существует по меньшей мере один не пустой входящий файл и на сервере Kafka не существует по меньшей мере одного объекта из входящих описаний объектов. В остальных случаях утилита будет говорить, что разницы нет.

    Если есть расхождения, то утилита выведет отличия во входящих файлах относительно кластера и отличия в кластере относительно входящих файлов.

    Утилита возвращает 0, если нет разницы; 1 — есть разница; 2 — если в вызове не были обнаружены валидные входящие объекты.

    Формат печати

    До версии 1.3.0 печать выполнялась в виде однострочных сортированных JSON-записей, которые тасовались друг с другом простым способом. С версии 1.3.0 такая печать заменена на многострочную, двухколоночную. Ниже показан фрагмент вывода.

    DIFF RESULT:
 There are differences in topics.
LEGEND:
 < [INPUT]
 > [ACTUAL]
----------------------------------
Differences in the fields: partitions
  {                                                                            | {
    "name" : "test1",                                                          |   "name" : "test1",
    "replicationFactor" : 1,                                                   |   "replicationFactor" : 1,
    "partitions" : 5,                                                          |   "partitions" : 2,
    "others" : {                                                               |   "others" : {
      "cleanup.policy" : "delete",                                             |     "cleanup.policy" : "delete",
      "min.insync.replicas" : "1",                                             |     "min.insync.replicas" : "1",
      "retention.ms" : "21600000"                                              |     "retention.ms" : "21600000"
    },                                                                         |   },
    "maxMessageSize" : 1048588                                                 |   "maxMessageSize" : 1048588
  }                                                                            | }

-----------------------------------------------------------------------------------------------------------------------------------------------------

  {                                                                            | {
    "name" : "test2",                                                          |   "comment" : "There is no comparable definition."
    "replicationFactor" : 1,                                                   | }
    "partitions" : -1,
    "others" : { },
    "maxMessageSize" : -1
  }

-----------------------------------------------------------------------------------------------------------------------------------------------------

  {                                                                            | {
    "comment" : "There is no comparable definition."                           |   "name" : "test1",
  }                                                                            |   "replicationFactor" : 1,
                                                                               |   "partitions" : 2,
                                                                               |   "others" : {
                                                                               |     "cleanup.policy" : "delete",
                                                                               |     "min.insync.replicas" : "1",
                                                                               |     "retention.ms" : "21600000"
                                                                               |   },
                                                                               |   "maxMessageSize" : 1048588
                                                                               | }

-----------------------------------------------------------------------------------------------------------------------------------------------------

 There are differences in ACLs.
LEGEND:
 < [INPUT]
 > [ACTUAL]
----------------------------------
  {                                                                            | {
    "resourceType" : "topic",                                                  |   "comment" : "There is no comparable definition."
    "mask" : "popic",                                                          | }
    "patternType" : "PREFIXED",
    "principal" : "User:bob",
    "operation" : "read",
    "host" : "*",
    "permissionType" : "ALLOW"
  }

-----------------------------------------------------------------------------------------------------------------------------------------------------

  {                                                                            | {
    "resourceType" : "topic",                                                  |   "comment" : "There is no comparable definition."
    "mask" : "test",                                                           | }
    "patternType" : "LITERAL",
    "principal" : "User:alice",
    "operation" : "describe",
    "host" : "*",
    "permissionType" : "ALLOW"
  }

-----------------------------------------------------------------------------------------------------------------------------------------------------

  {                                                                            | {
    "comment" : "There is no comparable definition."                           |   "resourceType" : "topic",
  }                                                                            |   "mask" : "test",
                                                                               |   "patternType" : "PREFIXED",
                                                                               |   "principal" : "User:alice",
                                                                               |   "operation" : "write",
                                                                               |   "host" : "*",
                                                                               |   "permissionType" : "ALLOW"
                                                                               | }

-----------------------------------------------------------------------------------------------------------------------------------------------------

    Вывод нужно понимать так:

    • В левой колонке выводится сериализованное в JSON представление объекта, которое определяется входящим YAML-файлом(и). В правой колонке выводится фактическое определение этого объекта на сервере Kafka.
    • Утилита выводит объекты всегда относительно входящей конфигурации в такой последовательности: сначала объекты, которые определены в конфигурации, потом объекты, которые есть на сервере, но которых нет в конфигурации.

    • Когда утилите не удается сопоставить объект, на пустом месте слева или справа выводится структура

      {
  "comment" : "There is no comparable definition."
}

      Если она появляется в правой колонке, это значит, что объекта с заявленным определением нет на сервере; если в левой — в конфигурации нет определения для существующего объекта.

    • Для топиков и ACL утилита делает сравнивание по всем полям структуры, которые она выводит.
    • Для топиков, в случаях когда есть сопоставление левого объекта с правым, перед структурой выводится строка Differences in the fields: <список-полей-через-запятую-в-которых-обнаружена-разница>.

    Режим валидации

    Примечание:
    Режим большой валидации работает только с форматом YAML. По этой причине поддержка формата CSV может быть убрана в будущих релизах.

    По умолчанию утилита исполняет конфигурацию так, как она передана пользователем с минимальными проверками корректности вхождений:

    • Для топиков минимальными требованиями являются:
      • непустой name;
      • непустой список permissions для CSV;
      • ненулевое значение для числа партиций;
      • ненулевое значение для фактора репликации;
      • непустой список allowed principals для CSV;
      • ненулевое значение для максимального размера сообщения в топике.
    • Для ACL минимальными проверками являются:
      • непустой тип объекта;
      • непустая маска ACL правила;
      • непустой список ограничений, причем все ограничения в списке должны быть допустимыми для данного типа объекта;
      • непустой список принципалов.

    Примечание:
    Минимальные проверки работают во всех форматах описания объектов.

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

    Режим большой валидации можно включить, передав опцию -V или --validate-before-execute, либо определив переменную окружения KOC__VALIDATE.

    Если режим валидации активирован, то включаются следующие проверки:

    1. Запрещаются дубликаты в списках. Эта проверка запрещает дубликаты в списках topics и acls. При обнаружении дубликатов, они отбрасываются из потока исполнения, а на терминал выводится предупреждение. В списке topics дубликатами считаются все определения в которых буквально совпадает имя топика. В acls дубликатами считаются элементы, в которых совпадают все поля, кроме поля disabled.
    2. Запускается проверка принципалов. Если в конфигурации определен словарь expectations и в нем есть непустой список principals, то валидатор бракует любой объект списка acls, в котором обнаруживается принципал, незарегистрированный в expectations.

    В режиме валидации утилита будет отправлять на исполнение валидные объекты, однако всегда будет возвращать код 1, если в потоке есть объекты, в которых проваливаются вышеперечисленные проверки.

    start-koc.sh

    Для запуска в окружении оболочки Bash, вы можете использовать вспомогательный сценарий start-koc.sh, идущий в поставке.

    ./start-koc.sh [-h] \
  [ -r | --replication-factor {number}] \
  [ --get-yaml ] \
  [ --no-print-replication-factor ] \
  [ --dir /path/to/dir ] \
  [ file1.yaml file2.yml file3.csv ...]

    Сценарий предполагает, что JAR-архив с утилитой лежит на одном с ним уровне, а диспозиция конфигурации — ./etc/config.properties. При вызове ОБЯЗАТЕЛЬНО нужно передать глобальный фактор репликации либо опцией -r, либо переменной окружения REPLICATION_FACTOR. Если фактор репликации определен в конфигурации, то значение из конфигурации перекроет значение, передаваемое сценарием, тем не менее, реализация сценария требует его определения также в вызове.

    Прочие переменные окружения:

    • JAVA[=java] — относительный или абсолютный путь до исполняемого файла JRE java.
    • CONF_DIR[=etc/config.properties] — путь до файла конфигурации утилиты.
    • JAVA_OPTS[='-Xmx512M -Xms512M'] — опции для JVM.
    • KOC__DIFF_MODE — переключатель в режим diff из переменной окружения. Чтобы переключатель заработал, необходимо присвоить переменной любое значение.
    • REPLICATION_FACTOR — глообальный фактор репликации, передаваемый переменной окружения. Перекрывается опцией -r.
    • KOC__BOOTSTRAP_SERVERS — позволяет передать сервер для подключения к кластеру Kafka из переменной окружения. Формат аналогичен тому, как если бы он передавался в файле конфигурации. Данный способ передачи параметра обладает самым низким приоритетом.

    В качестве аргументов вы можете передать один и более файлов с описанием объектов.

    Итоговые статистики

    В основном режиме запуска утилита завершает свою работу выводом итоговых статистик:

    TOTAL STATISTICS:
----------------
  Total processed lines . . . . . . . . . . . . . . . . . . : <счетчик>
  Total generated requests  . . . . . . . . . . . . . . . . : <счетчик> [batch_size: <размер>]
  Total ignored requests to create topic  . . . . . . . . . : <счетчик>
  Total ignored requests to delete topic  . . . . . . . . . : <счетчик>
  Total ignored requests to alter topic . . . . . . . . . . : <счетчик>
  Total ignored ACL requests  . . . . . . . . . . . . . . . : <счетчик>
  Total parsing errors  . . . . . . . . . . . . . . . . . . : <счетчик> [exps: <счетчик>]

# CЧЕТЧИКИ НИЖЕ ПОЯВЛЯЮТСЯ ТОЛЬКО В РЕЖИМЕ БОЛЬШОЙ ВАЛИДАЦИИ
  Total ignored topic duplicate lines . . . . . . . . . . . : <счетчик>

    Их нужно понимать так:

    • Total processed lines — для формата CSV этот счетчик увеличивается для каждой обработанной строки таблицы. Для YAML файла считается каждый обработанный элемент из списков topics и acls. Данный счетчик является общей суммой для всех переданных на вход файлов описаний. Этот счетчик учитывает как валидные, так и не валидные описания объектов.
    • Total generated requests — общее число сгенерированных запросов на основе всех валидных описаний объектов. Этот счетчик может быть не равен Total processed lines, даже когда все описания валидны. Это происходит потому, что команды для ACL описаний генерируются для каждого пользователя, хоста и операции во всех возможных комбинациях. Для топиков всегда генерируется по одной команде на создание/удаление, а также по команде на каждый изменяемый параметр. Рядом с этим счетчиком выводится использовавшийся размер пачки из полученных команд batch_size.
    • Total ignored requests to create topic — счетчик команд на создание топиков, которые были отброшены по причине того, что топики уже существуют.
    • Total ignored requests to delete topic — счетчик команд на удаление топиков, которые были отброшены по причине того, что топиков не существует.
    • Total ignored requests to alter topic — счетчик команд на изменение параметров топиков, которые были отброшены по некоторым причинам. Например, команда на изменение max.message.size в топике отбрасывается при попытке уменьшить значение для существующего топика.
    • Total ignored ACL requests — счетчик описаний ACL, которые были отброшены из-за логических ошибок. В режиме большой валидации этот счетчик подсчитывает также дубликаты описаний ACL.
    • Total parsing errors — счетчик невалидных описаний. Вычисляется как "Total processed lines" - "Correct lines" + "Exceptions". Под Correct lines понимаются такие описания, из которых может быть сгенерирован корректный запрос к серверу Kafka. Exceptions — это счетчик описаний, обработка которых привела к исключительной ситуации. Значение этого счетчика находится в квадратных скобках [exps:]. К исключительным ситуациям приводят ошибки парсера, в моменты когда он не может обработать некоторую часть внутри формата.
      Этот счетчик в хорошей ситуации должен быть нулевым, иначе следует обратить внимание на найденные ошибки, которые будут выводиться в логе сразу за статистиками.
    • Total ignored topic duplicate lines — счетчик режима большой валидации. Подсчитывает дубликаты описания топиков.

    Если во время просмотра элементов по спискам была найдена по меньшей мере одна логическая ошибка, утилита выводит лог ошибок сразу за статистиками.

    Режим опроса сервера

    Режим опроса сервера предназначен для выгрузки информации по существующим объектам на сервере Kafka и формирования на основе этих данных YAML файла. Этот режим может использоваться, например, если у вас есть сервер и вам нужно создать измененное описание объектов на основе тех объектов, которые имеет сервер.

    Для запуска этого режима необходимо указать опцию командной строки --get-yaml, либо передать переменную окружения KOC__GET_YAML с любым непустым значением.

    По умолчанию утилита сформирует файл в рабочем каталоге с именем koc-report-yyyy-MM-dd-HHmmss.yaml, где часть yyyy-MM-dd-HHmmss является временнОй меткой. Вы можете изменить директорию назначения результирующего файла, если укажете её через опцию --dir /путь/новой/директории. Директория должна существовать на момент запуска. Если директории не существует, либо переданный путь не является директорией, утилита не выведет ошибок, а тихо запишет файл в текущий рабочий каталог.

    В файл будут записаны следующие элементы:

    • список topics, если он не пустой;
    • список acls, если он не пустой;
    • список expectations, если список acls не пустой.

    Объекты в списках topics и acls выводятся в порядке, в котором их вернул сервер. Списки значений principals и operations всегда выводятся в сортированном порядке.

    По умолчанию YAML записывается в избыточной форме (с указанием всех возможных параметров), а для всех элементов списка topics выводится фактор репликации replicationFactor. Однако, иногда вам он может быть не нужен, так как его значение может передаваться из глобальных настроек утилиты. Чтобы подавить вывод этого параметра, используйте опцию --no-print-replication-factor в параметрах запуска утилиты, либо определите переменную окружения KOC__NO_REPLICATION_FACTOR и проинициализируйте её непустым значнием, либо укажите в конфигурации утилиты параметр koc.flag.no.print.replication.factor и присвойте ему любое непустое значение.

    Полученный на выходе YAML-файл соответствует формату 1.0.4 и может передаваться на вход утилите без каких-либо изменений.

    Логирование событий в файлы

    С версии 1.1.4 добавлено логирование некоторых событий в файлы. Логирование реализовано библиотекой Apache Log4J2. По умолчанию утилита логирует события уровня INFO на консоль. События уровня WARN дополнительно логируются в два файла:

    • ${log.dir}/koc-error-report.log — логирует все ошибки и предупреждения утилиты. В идеальной ситуации должен остаться пустым после отдельно взятого запуска утилиты, что означает отсутствие в конфигурации ошибок и успешность всех запросов к кластеру Kafka.
    • ${log.dir}/koc-report.log — файл с отчетом о проделанной работе. Логирует все запросы к серверу Kafka, в том числе и ошибки клиента Kafka. Некоторые события в этом файле отформатированы:
      • DELETE [<время> ms]: Requested topics: <список> — список топиков, для которых было запрошено удаление. Время в скобках показывает лаг между отправкой запроса и приемом ответа на него.
      • CREATE [<время> ms]: Requested topics: <список> — список топиков, для которых было запрошено создание. Время в скобках показывает лаг между отправкой запроса и приемом ответа на него.
      • ALTER [<время> ms]: Requested topics: <список> — список топиков, для которых было запрошено изменение параметров. Время в скобках показывает лаг между отправкой запроса и приемом ответа на него.
      • CREATE [<время> ms]: Requested ACLs: <список> — список ACL правил, для которых было запрошено создание. Время в скобках показывает лаг между отправкой запроса и приемом ответа на него.
      • DELETE [<время> ms]: Requested ACLs: <список> — список ACL правил, для которых было запрошено удаление. Время в скобках показывает лаг между отправкой запроса и приемом ответа на него.
      • Запросы клиента к серверу Kafka выполняются асинхронно. Для некоторых заданий утилита ожидает ответ сразу после отправки, для других утилита не ждет ответа, а выдает запросы другой категории (например, создание топиков и ACL правил не требует синхронизации). Когда алгоритм требует синхронизации с ответами, утилита целенаправленно ждет ответ на ранее отправленный запрос. Когда ответ получен, в лог выводится сообщение следующего содержания:
        • DONE #<счетчик> [<время> ms]: <строка> — результат обработки запроса. Счетчик отражает порядковый номер ответа в текущей стадии синхронного ожидания, а <время> отражает затраченное время на ожидание текущего ответа. В <строка> выводится ответ, сформированный сервером.

    Примечание:
    Файлы, которые создаются во время логирования, не ротируются, а перезаписываются, поэтому, если вам нужно сохранить результат запуска, перенесите или переименуйте файлы после завершения работы утилиты.

    По умолчанию утилита пишет логи в директорию ./koc-logs, но вы можете изменить диспозицию с помощью переменной окружения LOG_DIR. Если директории по переданному пути не существует, утилита предпримет попытку создать её.

    Конфигурация утилиты

    Конфигурационный файл утилиты хранит некоторые переменные для самой утилиты и переменные для подключения брокера. При чтении файла конфигурации утилита интерпретирует все известные ей параметры, а неизвестные параметры интерпретирует как параметры для подключения к Kafka серверу. Большинство параметров утилиты имеет значения по умолчанию.

    Ниже перечислены параметры самой утилиты и их описание:

    • csv.files — список файлов с описаниями объектов, разделенных символом ,. Данный параметр перекрывает любые другие способы объяления списка.
    • koc.csv.delimiter (по умолчанию ,) — разделитель полей в CSV файлах.
    • koc.csv.enclosure (по умолчанию ") — символ, используемый для отгораживания сложных значений в поле CSV (обычно, в которых есть пробелы).
    • koc.csv.escape (по умолчанию \) — символ, используемый для экранирования служебных символов в CSV.
    • koc.csv.batch.size_per_thread (по умолчанию 50) — размер одной пачки запросов, посылаемых одним потоком исполнения. (см. Особенности реализации)
    • replication.factor — глобальный фактор репликации. Аналогичен KOC__REPLICATION_FACTOR и -Dreplication.factor=, но всегда перекрывает их.

    Особенности реализации

    Утилита реализована в несколько потоков исполнения:

    • Парсер CSV файлов. Ориентируется на расширение файлов csv без учета регистра.
    • Парсер YAML файлов. Ориентируется на расширение файлов yml или yaml без учета регистра.
    • Поток исполнения заданий с периодом 500 мс вычитывает koc.csv.batch.size_per_thread запросов из списка и на их основе отправляет запросы брокеру Kafka. Поток педварительно сортирует поступающие задания по категориям, поэтому порядок их исполнения может быть не совсем таким, как он записан в файлах. Порядок запросов такой: сначала удаляются топики по списку, потом создаются топики по списку плюс для существующих топиков меняется максимальный размер сообщения, если есть на то запрос, затем создаются ACL по списку, потом удаляются ACL по списку. Порядок топиков и ACL объектов во всех списках соответствует порядку определения их в файлах и порядку вычитки каждого файла.

    Утилита останавливается, когда обработается последний запрос.

    С версии 1.0.5 добавлена проверка на существование объекта на сервере Kafka, что требует отправки запроса на выгрузку списков существующих объектов. Эта проверка отбрасывает задания по созданию уже существующих объектов (кроме исключения, описанного ниже) и удаление несуществующих, чтобы не нагружать сеть лишними запросами.

    Для случая, когда получено задание на создание существующего топика, проводится дополнительное действие, если заявленный и фактический размеры max.message.bytes для топика различаются:

    • Если фактический max.message.bytes меньше заявленного, то утилита сделает запрос на изменение фактического значения в большую сторону.
    • Если фактический max.message.bytes больше заявленного, утилита выдаст предупреждение на такой запрос, так как уменьшение максимального размера сообщения в существующем топике может сломать логику брокера и производителей, которые пишут в этот топик.

    Примечание:
    Apache Kafka не разрешает уменьшать количество партиций для существующих топиков, а также изменять фактор репликации на лету в автоматическом режиме, и эти действия также не поддерживаются утилитой. Другими словами, утилита не сможет применить новое значение фактора репликации и число партиций для существующего топика, однако, утилита позволяет пересоздать топик принудительно с теми параметрами, которые есть в описании объекта. Для этого следует использовать флаг recreate (по умолчанию false), чтобы дать подсказку утилите на случай, если топик уже существует. Будьте предельно осторожны выполняя пересоздание, так как топик может содержать сообщения, еще не прочитанные потребителями, которые будут потеряны.

    Форматы определения объектов в файлах

    Формат CSV

    Примечание:
    СSV является первым, старым форматом, с очень узкой направленностью. Предполагается, что YAML его заменит.

    Каждая стока представляет собой правило создания/удаления топика или ACL. Парсер CSV разрешает, чтобы таблица была не симметричной, т.е. число столбцов разных строк может быть не одинаковым.

    Общая структура строки такова:

    <идентификатор>,<имя>,<другие параметры> ...

    Идентификаторы, поддерживаемые реализацией:

    • TOPIC — создает топик.
    • RTOPIC — удаляет топик.
    • ACL — создает ACL для топика.
    • RACL — удаляет ACL для топика.
    • ACL_GROUP — создает ACL для группы.
    • RACL_GROUP — удаляет ACL для группы.
    • ACL_TID — создает ACL для транзакционного идентификатора.
    • RACL_TID — удаляет ACL для транзакционного идентификатора.

    Ниже приводятся параметры для каждого идентификатора:

    TOPIC,name,permissions[;...],partitions,replication,allow-principal[;...],,max.message.bytes
RTOPIC,name
[ACL|RACL|ACL_GROUP|RACL_GROUP],mask,permissions[;...],allow-principal[;...],additional
[ACL_TID,RACL_TID],mask,permissions[;...],allow-principal[;...]

    Примечание:
    Пустая 7-я ячейка в TOPIC является особенностью реализации и фактически значение из неё не берётся, но сама ячейка не может быть опущена (по историческим причинам).

    Описание полей:

    • name — имя топика в правилах, связанных с топиком. Имя всегда интерпретируется буквально.
    • mask — маска объектов в правилах с ACL. Маска всегда интерпретируется по шаблону LITERAL.
    • permissions — список ограничений, применяемых к объекту. Ограничения могут быть заданы списком: в этом случае их нужно разделять символом точки с запятой ;. Ограничения, передаваемые в тех или иных правилах, должны поддерживаться объектами.
    • allow-principals — список принципалов, для которых разрешаются переданные операции. Если принципалов несколько, то они могут быть переданы списком: в этом случае их нужно разделять символом точки с запятой ;. Формат CSV позволяет создавать только разрешающие правила.
    • partitions — количество партиций для создаваемого топика. Всегда должно быть больше 0.
    • replication — фактор репликации для создаваемого топика. Всегда должно быть больше 0.
    • max.message.bytes — размер сообщений в топике. Переопределяет значение по умолчанию. Всегда должно быть больше 0.

    Для ACL и ACL_GROUP можно использовать 5-е, необязательное поле additional. Оно используется, чтобы одновременно добавить правило для группы относительно топика для ACL и топик относительно группы для ACL_GROUP. Это позволяет свернуть некоторые правила. Например, следующее правило

    ACL,*,consumer;Describe,User:user1;User:user2,group-1

    сворачивает следующие два

    ACL,*,consumer;Describe,User:user1;User:user2
ACL_GROUP,group-1,consumer;Describe,User:user1;User:user2

    или аналогично такому вызову

    /opt/kafka/bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --topic "*" --group "group-1" --consumer --operation Describe --allow-principal "User:user1" --allow-principal "User:user2"

    т.е. определяется два правила в одном: для группы потребителей и для топика.

    Формат YAML

    Формат YAML лишен многих ограничений и позволяет создавать более полные правила. Общий вид формата версии 1.0.4 показан ниже. Некоторые поля были добавлены позднее первоначального формата. Для них сделана отметка в квадратных скобках, с какой версии формата поле появилось.

    # 1.0.5
---

#
# Примечание: все списки необязательны, но необходимо присутствие хотя бы одного из двух: topics или acl.
#

# КОЛЛЕКЦИЯ ЗНАЧЕНИЙ ПО УМОЛЧАНИЮ.
defaults:
    replication.factor: <число>      # (опционально) Положительное число. Фактор репликации по умолчанию для этой конфигурации.
    max.message.bytes: <число>       # (опционально) Положительное число. Максимальный размер сообщения топика для этой конфигурации.
    cleanup.policy: compact | delete # [1.0.3] (опционально) Допустимые значения 'compact' или 'delete'. Политика удаления устаревших сегментов топика. По умолчанию 'delete'.
    retention.ms: <число>            # [1.0.3] (опционально) Максимальный период удержания устаревших сегментов топика для политики 'delete'. По умолчанию 604800000 (7 дней).
    min.insync.replicas: <число>     # [1.0.3] Требование минимального числа живых реплик для партиции, когда производитель сообщений не указывает его явно. Значение не должно превышать значения фактора репликации. По умолчанию 1.

# СЛОВАРЬ ОЖИДАЕМЫХ ЗНАЧЕНИЙ
# [1.0.2]
expectations:
    principals: [ <expected-principal> ]    # [1.0.2] (опционально) По умолчанию null.

# СПИСОК ТОПИКОВ
topics:
    - name: topic-1                  # Имя топика. Должно быть уникальным для этого списка.
      disabled: false                # (опционально) По умолчанию false.
      recreate: false                # [1.0.2] (опционально) По умолчанию false.
      replication.factor: 3          # (опционально) Положительное число.
      partitions: 15                 # (опционально) Положительное число.
      others:                        
        max.message.bytes: 1000000   # (опционально) Положительное число. Размерность в байтах.
        cleanup.policy: compact | delete   # [1.0.3] (опционально) См. defaults.
        retention.ms: <число>              # [1.0.3] (опционально) См. defaults.
        min.insync.replicas: <число>       # [1.0.3] (опционально) См. defaults.

# ПРИМЕРЫ
    - name: topic-2
      replication.factor: 1
    - name: topic-3
      partitions: 10
    - name: topic-4
      others:
        max.message.bytes: 2000000

# СПИСОК ACL ПРАВИЛ
acls:
    - resource-type: topic           # Тип объекта: topic, group, transactional-id, cluster [1.0.4], delegation-token [1.0.4].
      mask: topic-1                  # Маска, по которой объекты указанного типа фильтруются.
      resource-pattern: LITERAL      # (опционально) Правило интерпретации маски: LITERAL (по умолчанию) или PREFIXED.
      principals: [ "User:alice", "User:john", "User:bob" ]    # Список строк. Список принципалов в формате 'User:<имя>'.
      operations: [ "read", "write", "describe" ]              # Список строк. Список операций разрешаемых/зпрещаемых для этого типа объектов.
      hosts: [ '*' ]                                           # Список строк. Хосты клинтов, на которые распространяются правила. По умолчанию ['*'].
      permission: ALLOW                                        # (опционально) Тип правила: ALLOW (по умолчанию) или DENY. ALLOW - разрешающее правило; DENY - запрещающее правило.

# ПРИМЕРЫ
    - resource-type: group
      mask: group-1
      resource-pattern: LITERAL
      principals: [ "User:bob" ]
      operations: [ "read" ]
      hosts: [ '*' ]
      permission: ALLOW
    - resource-type: transactional-id
      mask: '*'
      resource-pattern: LITERAL
      principals: [ "User:alice" ]
      operations: [ "describe" ]
      hosts: [ '*' ]
      permission: ALLOW

# [1.0.5] Объявление идентификаторов кластеров для прикрепления к ним некоторых объектов.
# Примечание: в 1.2.2 добавлена только в валидатор YAML, но по факту этот списое не используется.
clusters:
    - cluster-name-1
    - cluster-name-2

    Примечание:
    Особенности внутреннего сериализатора допускают написание имен параметров в некоторых местах спецификации, которые содержат точку или символ тире, в Camel Case нотации. В этом случае они будут интерпретироваться точно так же, как если бы писались по принятой здесь спецификации. Разрешается использовать Camel Case нотацию во всех местах, кроме параметров внутри defaults и others. Например, параметр фактора репликации допустимо писать как replication.factor, replication-factor и replicationFactor внутри элемента списка topics (будет работать одинаково). Но replication.factor внутри элемента defaults можно писать только с точкой. В режиме --get-yaml утилита всегда использует Camel Case, потому что она сериализует из POJO-объекта, где нельзя использовать точки и тире в именах атрибутов без усложнения сериализатора.

    Описание формата 1.0.5

    • Добавлен список:
      • clusters — (поддержка запланирована в одном из будущих релизов) объявляет идентификаторы кластеров, за которыми можно закрепить несколько объектов в списках для частичного отключения определений в части спецификации.

    Описание формата 1.0.4

    • Добавлена поддержка значений:

      • acls

        • Элемент списка:

          • resource-type:

            • cluster — дает права на администрирование параметров Kafka сервера, в том числе хранилища с мета-данными.

              Примечание:
              Тип ресурса cluster поддерживает для поля mask всего одно значение — kafka-cluster. Таким образом, определение ACL для этого ресурса всегда начинается так:

              acls:
- resource-type: cluster
  mask: kafka-cluster
  ...
            • delegation-token — дает права на просмотр делегирующих токенов механизма аутентификации Delegation Token API.

    Описание формата 1.0.3

    Примечание:
    Вводимые значения для параметров не проверяются на корректность и передаются на сервер транзитом (валидация на стороне сервера Apache Kafka). Параметры участвуют в формировании запросов для создающих правил, в том числе и для ситуаций, когда топик существует.

    • Дополнения:
      • defaults — поддержка значений по умолчанию для конфигурации YAML для параметров топиков:
        • cleanup.policy — политика удаления устаревших сегментов топика;
        • retention.ms — максимальный период удержания устаревших сегментов топика для политики delete;
        • min.insync.replicas — требование минимального числа живых реплик для партиции, когда производитель сообщений не указывает его явно.
      • topics
        • Элемент списка:
          • others
            • cleanup.policy — если не указан явно для топика, то берется из defaults. Если нет значения в defaults, то используется настройка сервера Apache Kafka.
            • retention.ms — если не указан явно для топика, то берется из defaults. Если нет значения в defaults, то используется настройка сервера Apache Kafka.
            • min.insync.replicas — если не указан явно для топика, то берется из defaults. Если нет значения в defaults, то используется настройка сервера Apache Kafka.

    Описание формата 1.0.2

    • Дополнения:
      • expectations — необязательная коллекция ожидаемых значений. Используется для валидации YAML файла. Хранит ожидаемые значения, от которых пользователь не может отклоняться в определениях объектов. Данная коллекция имеет область видимости, ограниченную структурой YAML, в которой она объявлена.
        • principals — список принципалов, которые можно использовать в определениях объектов списка acls. Используется только если он не пустой валидаторами структуры YAML. Если этот список не пустой, то каждый принципал из определения объекта списка acls будет проверен на пристутствие его в списке expectations, и если принципала в нем нет, то он будет отброшен из правила.
      • topics
        • Элемент списка:
          • recreate (по умолчанию false) — дает утилите инструкцию тихо удалить топик и создать его с заявленными параметрами. Данный флаг работает только в случае, когда дается команда на создание топика; в остальных случаях игнорируется независимо от значения.

    Описание формата 1.0.1

    • defaults — коллекция значений по умолчанию, которые применяются тогда, когда они не определены явно в объектах. Данная коллекция имеет область видимости, ограниченную структурой YAML, в которой она объявлена.
      • replication.factor — глобальный фактор репликации. Будет использоваться по умолчанию, если он не был определен в вызове утилиты через командную строку.
      • max.message.bytes — глобальный размер сообщения в топике.
    • topics — список топиков, которые нужно создать/удалить.
      • Элемент списка:
        • name — обязательный параметр; определяет имя топика.
        • disabled (по умолчанию false) — определяет должен ли топик быть создан (false) или удалён (true).
        • replication.factor — задает фактор репликации для топика. Всегда должен быть больше нуля. Если не определен здесь, то проверяется значение из defaults. Если и там значение не определено, то проверяется конфигурация приложения. В текущей реализации этот параметр не может быть не задан на всех возможных уровнях.
        • partitions — необязательный параметр; определяет число партиций для создаваемого топика.
        • others — коллекция прочих параметров топика.
          • max.message.bytes — максимальный размер сообщения топика в байтах. Если не указан здесь, будет предпринята попытка найти значение в defaults. Если и там нет, будет использоваться значение по умолчанию в кластере Kafka.
    • acls — список ACL правил, которые нужно создать/удалить.
      • Элемент списка:
        • disabled (по умолчанию false) — определяет должен ли ACL быть создан (false) или удалён (true).
        • resource-type — обязательный параметр; определяет тип ресурса, для которого создается ACL: topic, group или transactional-id.
        • mask — обязательный параметр; определяет имя одного или множества объектов, в зависимости от шаблона.
        • resource-pattern (по умолчанию LITERAL) — шаблон, определяющий метод интерпретации mask. Поддерживаются LITERAL и PREFIXED. Значение этого параметра чувствительно к регистру: допустим только верхний регистр.
        • principals — список принципалов, на которые распространяется данное правило. Допустимо передавать пустой список: в этом случае правило не ограничивается принципалами.
        • operations — список операций, которые ограничивают принципалов в отношении данного объекта. Элементы списка зависят от типа ресурса (см. документацию Kafka). Допустимо передавать пустой список: в этом случае правило разрешит все допустимые операции.
        • hosts — список хостов, на которые распространяется указанное правило. Допустимо передавать пустой список: в этом случае всегда предполагается значение [ '*' ].
        • permission (по умолчанию ALLOW) — определяет разрешающее или запрещающее это правило. Поддерживается два значения: ALLOW и DENY. Значения не чувствительны к регистру.

    Правила обработки YAML файлов

    При обработке файлов YAML, парсер использует следующие правила и соглашения:

    • Далее под структурой мы понимаем валидное описание объектов Kafka в формате YAML, свойством которой является неделимость. Неделимая структура — неделимая единица — может храниться в одном файле или являться подструктурой внутри файла в Multi-YAML режиме.
    • Все файлы формата YAML подаются парсеру в порядке, передаваемом командной строкой. Каждый файл обрабатывается независимо от других.
    • Утилита не требует обязательно прописывать --- для обозначения начала структуры, тем не менее, в Multi-YAML режиме признак YAML должен быть перед началом каждой структуры.

    • Парсер понимает Multi-YAML режим с версии 1.1.0. Это позволяет вам передать несколько структур внутри одного файла (когда это удобно), при этом они будут обработаны так, как будто они записаны разными файлами, например

      # Пример Multi-YAML
---
# Первая структура
topics:
- name: topic-1
- name: topic-2
---
# Вторая структура
topics:
- name: topic-3
- name: topic-4
- name: topic-5

      Такой способ подачи может использоваться, если несколько YAML файлов предварительно конкатенируются:

      cat 1.yaml 2.yaml 3.yaml > big.yaml
./start-koc.sh <...> big.yaml
    • defaults, expectations, topics и acls являются элементами верхнего уровня. Ни один из них не является обязательным, но необходимо наличие хотя бы одного непустого из них, чтобы структура считалась валидной. При отсутствии topics и acls в валидной структуре просто не будет сгенерировано ни одного запроса. Парсер допускает дубликаты элементов верхнего уровня в рамках одной структуры, но при этом всегда будет использоваться только последнее вхождение.
    • Структура проходит несколько стадий проверок: сначала она проверяется на возможность десериализации её в задаваемый формат (мэпинг), а потом валидируется каждый элемент списка (минимальная валидация и большая валидация) с попутной генерацией команд. Парсер обрабатывает структуру всегда в один проход и не возвращается к уже обработанным элементам.

    Известные ограничения утилиты и рекомендации

    В этом разделе приводятся рекомендации по использованию утилиты, чтобы предупредить возможные ошибки.

    • Всегда используйте только административную учетную запись при подключении к Kafka-брокеру.
      Если на сервере Kafka настроена авторизация и выставлен параметр allow.everyone.if.no.acl.found=false, то утилита при запросе топиков и ACL списков будет получать только ту выборку, к которой пользователь, от которого утилита вызывается, имеет доступ в соответствии с текущим ACL на сервере. Административная учетная запись всегда игнорирует ACL, поэтому запросы от её имени всегда возвращают всю имеющуюся информацию. Также не используйте авторизацию через механизм PLAINTEXT, потому что она всегда делается от имени пользователя User:ANONYMOUS, у которого по умолчанию нет ACL прав. Если вы не будете использовать админстративную учетную запись, то возможны следующие ошибки:
      • При создании объектов может быть нарушен механизм проверки существующих объектов, в частности, утилита может попытаться создать существующий объект, потому что запрос на выгрузку не вернул его из-за ACL.
      • Режим верификации будет всегда работать некорректно при неполной выборке, что может приводить к ложным выводам.

    • При составлении файлов описаний объектов, нужно мыслить декларативно.
      Хотя CSV и YAML формат обрабатываются в поточном режиме, при их составлении нужно думать, что они декларируют состояние сервера. Другими словами, конфигурация говорит, какими объектами должен обладать / не должен содержать серевер. Таким образом, декларативный стиль требует соблюдения следующих правил:

      • Не описывайте один и тот же объект несколькими элементами списка.

        # НЕПРАВИЛЬНО
topics:
- name: topic-1
 partitions: 15
 disabled: true
- name: topic-1
 partitions: 30

        Предыдущий пример описывает объект не в декларативном стиле. В режиме запуска -V утилита не пропустит вторую запись, так как она дублирует первую и топик не будет создан. Так как данное описание стремится создать топик с 30 партициями, то нужно оставить только вторую запись:

        # ПРАВИЛЬНО
topics:
- name: topic-1
 partitions: 30

        Примечание:
        Если топик topic-1 уже существует и вы хотите увеличить ему число партиций, единственный способ сделать это через утилиту, это пересоздать его:

        topics:
- name: topic-1
  recreate: true
  partitions: 30

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

      • Не злоупотребляйте общими ACL правилами.
        Маскирующие символы передаются утилитой транзитом и обрабатываются только сервером Kafka. Другими словами, утилита не может автоматически обнаружить дубликаты после обработки маски. При попадании логических дубликатов в одну пачку запросов, могут получаться неожиданные результаты. Следующий пример наглядно демонстрирует логическое дублирование.

        # НЕПРАВИЛЬНО
acls:
- resource-type: topic
 mask: '*'
 operations:
  - all
 principals:
  - User:alice
  - User:bob
 disabled: true
- resource-type: topic
 mask: '*'
 operations:
  - alter
  - read
  - write
  - create
  - describe
  - describeconfigs
 principals:
  - User:alice
  - User:bob

        Предыдущий пример описывает два ACL-правила: два правила вводятся для одного и того же класса объектов для одних и тех же хостов и пользователей, при этом первое правило отбирает все права, а второе правило добавляет некоторые из них. При попадании этих правил в одну пачку запросов, сервер Kafka применит только первое правило, потому что оно более общее с точки зрения брокера при такой постановке. Чтобы исправить ситуацию, нужно чтобы запрещаемые и разрешаемые операции в правилах не пересекались. Для топика допустимыми являются alter, alterconfigs, create, delete, describe, describeconfigs, read, write, тогда устранить пересечение можно, указав в первом правиле delete и alterconfigs, а оставшиеся во втором:

        # ПРАВИЛЬНО
acls:
- resource-type: topic
 mask: '*'
 operations:
  - delete
  - alterconfigs
 principals:
  - User:alice
  - User:bob
 disabled: true
- resource-type: topic
 mask: '*'
 operations:
  - alter
  - read
  - write
  - create
  - describe
  - describeconfigs
 principals:
  - User:alice
  - User:bob

        Если правил до этого не существовало на сервере, то в такой постановке рекомендуется оставить только второе, создающее описание.

        Примечание:
        На практике следует отдавать предпочтение конкретным правилам, в которых в mask имя указывается конкретно, либо маскируется только его часть, например, topic-*.

        Приложения

        Таблица соответствия ACL операций

        resource-type operation
        topic all, alter, alterconfigs, create, delete, describe, describeconfigs, read, write, idempotentwrite, producer, consumer
        group all, delete, describe, read, idempotentwrite, consumer
        transactional-id all, describe, write, idempotentwrite
        cluster all, alter, alterconfigs, clusteraction, create, describe, describeconfigs, idempotentwrite
        delegation-token all, describe

        Операции producer и consumer это несколько свернутых операций.

        resource-type operation Сворачиваются
        topic producer write, describe, create
        topic consumer read, describe
        group consumer read
    Описание

    Java
    Readme
    0 Форки
    2 Звезд
    0 Наблюдатели
    Конвейеры
    0 успешных
    0 с ошибкой