Управление топиками через Kafka CLI^Beta

С помощью Kafka CLI можно создать топик Cloud Kafka, посмотреть его параметры, изменить настройки и удалить топик.

Доступные действия с топиками зависят от роли, которую назначают пользователю Cloud Kafka при создании экземпляра или добавлении учетной записи:

• Владелец кластера: может создавать топики, просматривать их параметры, увеличивать количество партиций, изменять срок хранения сообщений и удалять топики.
• Стандартный пользователь: может создавать топики, просматривать их параметры, увеличивать количество партиций, изменять срок хранения сообщений и удалять топики.
• Наблюдатель и Аудитор безопасности: могут просматривать список топиков и их параметры.

1. Подключитесь к экземпляру сервиса через Kafka CLI.

2. Создайте топик с нужным количеством партиций и фактором репликации:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --create \    --topic <ИМЯ_ТОПИКА> \    --partitions <КОЛИЧЕСТВО_ПАРТИЦИЙ> \    --replication-factor <ФАКТОР_РЕПЛИКАЦИИ>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --create `    --topic <ИМЯ_ТОПИКА> `    --partitions <КОЛИЧЕСТВО_ПАРТИЦИЙ> `    --replication-factor <ФАКТОР_РЕПЛИКАЦИИ>

   Здесь:

   • <ИМЯ_ТОПИКА> — имя создаваемого топика, например test-topic.
   • <КОЛИЧЕСТВО_ПАРТИЦИЙ> — нужное количество партиций.
   • <ФАКТОР_РЕПЛИКАЦИИ> — количество копий каждой партиции. Значение не может превышать количество узлов-брокеров в экземпляре сервиса. Их количество можно посмотреть на вкладке Архитектура страницы экземпляра сервиса.

   После создания топика в терминале появится сообщение: Created topic <ИМЯ_ТОПИКА>..

3. Проверьте, что топик появился в списке.

1. Подключитесь к экземпляру сервиса через Kafka CLI.

2. Получите список топиков:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --list

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --list

3. Посмотрите параметры нужного топика:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --describe \    --topic <ИМЯ_ТОПИКА>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --describe `    --topic <ИМЯ_ТОПИКА>

   Здесь <ИМЯ_ТОПИКА> — имя топика из полученного списка.

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

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

1. Подключитесь к экземпляру сервиса через Kafka CLI.

2. Посмотрите текущее количество партиций топика:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --describe \    --topic <ИМЯ_ТОПИКА>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --describe `    --topic <ИМЯ_ТОПИКА>

   Здесь <ИМЯ_ТОПИКА> — имя изменяемого топика.

   Текущее количество партиций будет указано в поле PartitionCount.

3. Укажите новое общее количество партиций, которое больше текущего:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --alter \    --topic <ИМЯ_ТОПИКА> \    --partitions <НОВОЕ_КОЛИЧЕСТВО_ПАРТИЦИЙ>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --alter `    --topic <ИМЯ_ТОПИКА> `    --partitions <НОВОЕ_КОЛИЧЕСТВО_ПАРТИЦИЙ>

   Здесь <НОВОЕ_КОЛИЧЕСТВО_ПАРТИЦИЙ> — общее количество партиций после изменения.

4. Проверьте, что значение PartitionCount совпадает с указанным новым количеством:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --describe \    --topic <ИМЯ_ТОПИКА>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --describe `    --topic <ИМЯ_ТОПИКА>

Параметр retention.ms задает срок хранения сообщений в миллисекундах, если cleanup.policy имеет значение delete или delete,compact. По умолчанию в Cloud Kafka используется delete, а срок хранения не ограничен (retention.ms=-1).

1. Подключитесь к экземпляру сервиса через Kafka CLI.

2. Установите нужный срок хранения сообщений:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-configs" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --alter \    --entity-type topics \    --entity-name <ИМЯ_ТОПИКА> \    --add-config retention.ms=<СРОК_ХРАНЕНИЯ_МС>

    
   & "$env:K_CLI_BASE\kafka-configs.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --alter `    --entity-type topics `    --entity-name <ИМЯ_ТОПИКА> `    --add-config retention.ms=<СРОК_ХРАНЕНИЯ_МС>

   Здесь:

   • <ИМЯ_ТОПИКА> — имя изменяемого топика.
   • <СРОК_ХРАНЕНИЯ_МС> — срок хранения сообщений в миллисекундах. Например, 86400000 — одни сутки.

   После изменения срока хранения в терминале появится сообщение: Completed updating config for topic <ИМЯ_ТОПИКА>.

3. Проверьте значение параметра:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-configs" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --describe \    --entity-type topics \    --entity-name <ИМЯ_ТОПИКА>

    
   & "$env:K_CLI_BASE\kafka-configs.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --describe `    --entity-type topics `    --entity-name <ИМЯ_ТОПИКА>

   В выводе значение retention.ms должно совпадать с указанным сроком хранения.

1. Подключитесь к экземпляру сервиса через Kafka CLI.

2. Удалите топик:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --delete \    --topic <ИМЯ_ТОПИКА>

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --delete `    --topic <ИМЯ_ТОПИКА>

   Здесь <ИМЯ_ТОПИКА> — имя удаляемого топика.

3. Получите список топиков и проверьте, что удаленного топика в нем нет:

   Linux/macOS

   Windows

    
   "${K_CLI_BASE}/kafka-topics" \    --command-config "${K_CONFIG}" \    --bootstrap-server "${K_BOOTSTRAP}" \    --list

    
   & "$env:K_CLI_BASE\kafka-topics.bat" `    --command-config "$env:K_CONFIG" `    --bootstrap-server "$env:K_BOOTSTRAP" `    --list