Автоматическое переподключение неудачных задач в Kafka-Connect

Question

Я использую плагин mongo-source с Kafka-connect. Я проверил состояние исходной задачи, и она работала и слушала коллекцию mongo.

Я вручную остановил службу mongod и подождал около 1 минуты, а затем снова запустил ее.

Я проверил исходную задачу, чтобы увидеть, исправится ли что-нибудь само, и через 30 минут кажется, что ничего не работает.

Только после перезапуска коннектора снова заработал.

Поскольку mongo-source не имеет параметров для установки повторных попыток + отсрочки при тайм-ауте, я искал конфигурацию, которая соответствовала бы простому сценарию: перезапустить неудачную задачу после X времени, используя конфигурацию Kafka-connect. не смог найти..:/ Я могу сделать это с помощью простого скрипта, но в Kafka-connect должно быть что-то, что управляет неудачными задачами. или даже в mongo-source... Я не хочу, чтобы он так быстро выходил из строя всего через 1 минуту...:/

Answer 1

Нет никакого другого способа, кроме использования REST API, чтобы найти сбойную задачу и отправить запрос на перезапуск, а затем запускать его на периодической основе. Например

curl -s "http://localhost:8083/connectors?expand=status" | \
  jq -c -M 'map({name: .status.name } +  {tasks: .status.tasks}) | .[] | {task: ((.tasks[]) + {name: .name})}  | select(.task.state=="FAILED") | {name: .task.name, task_id: .task.id|tostring} | ("/connectors/"+ .name + "/tasks/" + .task_id + "/restart")' | \
  xargs -I{connector_and_task} curl -v -X POST "http://localhost:8083"\{connector_and_task\}

Источник: https://rmoff.net/2019/06/06/automatically-restarting-failed-kafka-connect-tasks/