我在flink1.11.1中编写了一个flink批处理作业。作业成功完成后，我想做一些事情，比如调用http服务。
我添加了一个简单的作业侦听器来钩住作业状态。问题是当kafka sink操作符抛出错误时，不会触发job listener。我希望当我的作业失败时，它会触发我的作业侦听器并打印失败日志。
我怎样才能确定这项工作成功与否？
任何帮助都将不胜感激。
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)
1条答案
按热度按时间b1uwtaje1#
如果您尝试在群集上运行作业，可以在控制台中使用作业id查看日志记录消息和标准输出。请参阅所附的屏幕截图，
默认url可以是http://localhost：8081如果您在本地群集上运行。
同样，下面的方法是不正确的检查你的工作是否成功。