JavaNotes

Команды Kafka

2 min read

Команды Kafka

Команды ниже используют нейтральный локальный адрес. В рабочем окружении endpoint, TLS/SASL и файл клиентской конфигурации приходят извне; секреты не передаю прямо в командной строке.

export BOOTSTRAP_SERVER=localhost:9092
export TOPIC_NAME=orders
export GROUP_NAME=billing

Топики

kafka-topics --bootstrap-server "$BOOTSTRAP_SERVER" \
  --create --topic "$TOPIC_NAME" \
  --partitions 3 --replication-factor 1
 
kafka-topics --bootstrap-server "$BOOTSTRAP_SERVER" --list
 
kafka-topics --bootstrap-server "$BOOTSTRAP_SERVER" \
  --describe --topic "$TOPIC_NAME"

Увеличить число партиций можно, уменьшить — нельзя:

kafka-topics --bootstrap-server "$BOOTSTRAP_SERVER" \
  --alter --topic "$TOPIC_NAME" --partitions 6

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

Отправка и чтение

kafka-console-producer \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --topic "$TOPIC_NAME"
 
kafka-console-consumer \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --topic "$TOPIC_NAME" \
  --from-beginning

--from-beginning влияет на чтение при отсутствии подходящего committed offset; он не “перематывает” существующую группу сам по себе. Для наблюдения за конкретной группой:

kafka-console-consumer \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --topic "$TOPIC_NAME" \
  --group "$GROUP_NAME"
 
kafka-consumer-groups \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --describe --group "$GROUP_NAME"

Сброс offset

Сначала останавливаю потребителей группы и делаю dry run:

kafka-consumer-groups \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --group "$GROUP_NAME" \
  --topic "$TOPIC_NAME" \
  --reset-offsets --to-earliest --dry-run

Только после проверки результата заменяю --dry-run на --execute. Сброс способен повторно запустить бизнес-операции, поэтому до него проверяю идемпотентность.

Конфигурация защищённого клиента

kafka-topics \
  --bootstrap-server "$BOOTSTRAP_SERVER" \
  --command-config ./client.properties \
  --list

Сам client.properties не коммичу, если в нём есть credentials. В репозитории допустим шаблон без значений.

Типичные сбои

  • Запустить destructive-команду не на том кластере из-за старой переменной.
  • Считать replication factor 1 приемлемым для отказоустойчивого production.
  • Сбросить offset при работающей группе и получить конфликт или неожиданный rebalance.
  • Удалить топик вместо очистки тестовых данных.
  • Передать пароль параметром и оставить его в shell history.

Вопросы для собеседования

Как увидеть consumer lag? Описать consumer group и сравнить current offset с log end offset по партициям.

Почему --from-beginning иногда не читает старые сообщения? У группы уже есть committed offsets.

Можно ли уменьшить число партиций командой alter? Нет; обычно создают новый топик и мигрируют поток.

Зачем сначала --dry-run при reset offsets? Увидеть новые позиции до изменения состояния группы.

См. Kafka, Kafka Broker, Кейс конфигурация Kafka в разных окружениях.

Graph View

Backlinks