问题描述
我在flink 1.11.1中编写了一个flink批处理作业。工作成功完成后,我想执行一些操作,例如调用http服务。
我添加了一个简单的作业侦听器以挂钩作业状态。问题是当卡夫卡接收器运算符抛出错误时,作业监听器不会触发。我希望当我的工作失败时,它会触发我的工作监听器并打印失败日志。
如何确定作业是否成功完成?
任何帮助将不胜感激。
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.registerJobListener(new JobListener {
override def onJobSubmitted(jobClient: JobClient,throwable: Throwable): Unit = {
if (throwable == null) {
log.info("SUCCESS")
} else {
log.info("FAIL")
}
}
override def onJobExecuted(jobExecutionResult: JobExecutionResult,throwable: Throwable): Unit = {
if (throwable == null) {
log.info("SUCCESS")
} else {
log.info("FAIL")
}
}
})
env.createInput(input)
.filter(r => Option(r.token).getorElse("").nonEmpty)
.addSink(kafkaProducer)