$ command-helper
enru

Apache Kafka

Топики, продюсер и консьюмер из консоли, группы консьюмеров, lag и сброс оффсетов.

Потребуется вход через Google

Топики

Список топиков
kafka-topics.sh --bootstrap-server <broker> --list
Описание топика (партиции, реплики, ISR)
kafka-topics.sh --bootstrap-server <broker> --describe --topic <topic>
Создать топик
kafka-topics.sh --bootstrap-server <broker> --create --topic <topic> --partitions 3 --replication-factor 3
Увеличить число партицийпродвинутая
kafka-topics.sh --bootstrap-server <broker> --alter --topic <topic> --partitions 6

Уменьшить число партиций нельзя. Порядок сообщений по ключу после изменения нарушится.

Удалить топик
kafka-topics.sh --bootstrap-server <broker> --delete --topic <topic>
Партиции без полного набора репликпродвинутая
kafka-topics.sh --bootstrap-server <broker> --describe --under-replicated-partitions

Продюсер и консьюмер

Прочитать топик с начала
kafka-console-consumer.sh --bootstrap-server <broker> --topic <topic> --from-beginning
Прочитать с ключами, партициями и временем
kafka-console-consumer.sh --bootstrap-server <broker> --topic <topic> --property print.key=true --property print.partition=true --property print.timestamp=true
Прочитать N сообщений и выйти
kafka-console-consumer.sh --bootstrap-server <broker> --topic <topic> --from-beginning --max-messages 10
Отправить сообщения (каждая строка — сообщение)
kafka-console-producer.sh --bootstrap-server <broker> --topic <topic>
Отправить сообщения с ключом
kafka-console-producer.sh --bootstrap-server <broker> --topic <topic> --property parse.key=true --property key.separator=:

Группы консьюмеров

Список групп
kafka-consumer-groups.sh --bootstrap-server <broker> --list
Lag группы по партициям
kafka-consumer-groups.sh --bootstrap-server <broker> --describe --group <group>
Участники группы
kafka-consumer-groups.sh --bootstrap-server <broker> --describe --group <group> --members
Сбросить оффсеты на начало (dry run)продвинутая
kafka-consumer-groups.sh --bootstrap-server <broker> --group <group> --topic <topic> --reset-offsets --to-earliest --dry-run

Для применения замените --dry-run на --execute. Группа должна быть остановлена.

Сбросить оффсеты на момент временипродвинутая
kafka-consumer-groups.sh --bootstrap-server <broker> --group <group> --topic <topic> --reset-offsets --to-datetime 2024-01-01T00:00:00.000 --execute

Конфигурация

Конфиг топика
kafka-configs.sh --bootstrap-server <broker> --entity-type topics --entity-name <topic> --describe
Изменить retention топика (7 дней)продвинутая
kafka-configs.sh --bootstrap-server <broker> --entity-type topics --entity-name <topic> --alter --add-config retention.ms=604800000
Последние оффсеты по партициямпродвинутая
kafka-get-offsets.sh --bootstrap-server <broker> --topic <topic>

В версиях Kafka до 3.0 используйте kafka-run-class.sh kafka.tools.GetOffsetShell.