Если конвейер останавливается из-за неудачного теста или неисправных записей, необходимо сбросить офсет(смещение) кустомера (потребителя) или группы потребителей.
Будь то неудачный тестовый пример, дефектные записи данных или ошибка в самом потребляющем приложении: если потребитель не продолжает работу, конвейер данных останавливается, и все последующие процессы больше не получают данных. В таком случае есть только один вариант: записи данных, которые привели к проблеме, должны быть пропущены.
Для этого можно использовать консольное приложение, поставляемое вместе с Apache Kafka, при условии, что у вас есть прямой доступ к кластеру и соответствующие права.
Как узнать, какой кустомер завис?
Если вы уже нашли своего проблемного ребенка, то можете смело пропустить этот раздел.
Начиная с версии 2.4.0 Apache Kafka, появилась возможность проверить список всех групп потребителей и их отставание через консольное приложение "kafka-consumer-groups". Команда для этого следующая:
| 1 | kafka-consumer-groups --bootstrap-server localhost:9092 --all-groups -describe |
Если вы еще не используете версию 2.4.0 или более позднюю или не имеете доступа через консольное приложение, дополнение --all-groups недоступно, и вам следует полагаться на сторонние приложения для эффективного решения этого сценария использования.
Если вам уже известна группа потребителей, вы можете слегка модифицировать приведенную выше команду для получения lag известной группы потребителей (в примере: "myConsumerGroup"):
| 1 | ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group myConsumerGroup --describe |
Сброс смещений потребителей
После того, как вы нашли потребителя или группу потребителей, которые вы хотите сбросить, вы должны убедиться, что группа потребителей не активна: все приложения в группе потребителей должны быть выключены. В противном случае смещения потребителя или группы потребителей не могут быть изменены.
Используя консольное приложение "kafka-consumer-groups", сбросьте смещения следующим образом:
| 1 | ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group [myConsumerGroup] --reset-offsets --to-earliest --topic [my_topic] -execute |
Вместо --to-earliest, которая заставляет потребителя вернуться к началу, --to-latest может быть использована для установки потребителя в конец потока.
| 1 | ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group [myConsumerGroup] --reset-offsets --to-latest --topic [my_topic] -execute |
Таким образом, все сообщения, не потребленные до сих пор, будут пропущены. В качестве альтернативы есть возможность указать время:
| 1 | ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group [myConsumerGroup] --reset-offsets --to-datetime 2020-12-20T00:00:00.000 --topic [my_topic] --execute |
Заключение
В этой небольшой статье блога мы показали, как управлять смещениями потребителя с помощью консольного приложения, поставляемого с Apache Kafka.