Я пытаюсь выполнить одно действие после завершения задания flink (внесите некоторые изменения в базу данных).Я хочу сделать это в том же приложении Flink без удачи.Я обнаружил, что есть JobStatusListener, который уведомлен в ExecutionGraph об измененном состоянии, но я не могу найти, как я могу получить этот ExecutionGraph для регистрации моего слушателя.Я попытался полностью заменить ExecutionGraph в моем проекте (да, плохой подход, но ...), но как только это библиотека времени выполнения, она вообще не вызывается в распределенном режиме, только при локальном запуске.
У меня есть следующее короткое заявление:
DataSource.output(RichOutputFormat.class)
ExecutionEnvironment.getExecutionEnvironment().execute()
Может кто-нибудь помочь?