Как управлять оффсетом топика¶

Иногда появляется необходимость изменить текущий оффсет топика. Например, чтобы обработать его заново, или, наоборот, разгрести большую кучу, чтобы побыстрее привести систему в актуальное состояние.
Вот, что для этого нужно сделать (здесь и далее примеры с использованием этой функции-обёртки, вместо плейсхолдеров, окружённых %, подставить нужные значения)

1. Останавливаем все консьюмеры из нужной группы. Смысл в том, что, пока есть активные консьюмеры, команды изменения оффсета выполнять запрещено;
2. пишем команду kfk consumer-groups --describe --group %group_name% — она покажет нам все топики, на которые подписана группа, их оффсеты и лаги;
3. пишем команду
   kfk consumer-groups --group %group_name% --reset-offsets --to-latest --execute --topic %topic_name%
4. снова пишем kfk consumer-groups --describe --group %group_name% — убеждаемся, что в интересующем нас топике лаг сброшен в ноль;
5. запускаем консьюмеры.

Параметры¶

• --all-groups — применить ко всем группам консьюмеров (все группы должны быть без активных консьюмеров);
• --all-topics — применить ко всем топикам группы;
• --by-duration— сдвинуть на временной интервал относительно текущего времени (странный формат строки PnDTnHnMnS);
• --dry-run — показывает результат выполнения команды, не выполняя её;
• --to-datetime — сдвинуть на дату;
• --to-earliest — на самый ранний;
• --to-latest — на самый поздний (то, что использовано в примере);
• --shift-by — сместить оффсет назад (отрицательное число) или вперёд (положительное);
• --from-file — взять значения оффсета, куда сместить, из csv файла;
• --to-current — на текущий.

Ссылки¶

https://www.conduktor.io/kafka/kafka-consumer-group-management-cli-tutorial
https://kafka.apache.org/documentation/basic_ops_consumer_group

Комментарии

Комментарии