非常感谢菲利佩。这正是我所需要的，多亏了你，一切都完成了。我在这里共享代码，以防其他人需要。
让听众做好准备

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(...);
final AtomicReference<JobID> jobIdReference = new AtomicReference<>();
//Environment configurations
env.registerJobListener(new JobListener() {
    @Override
    public void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable) {
        assert jobClient != null;
        jobIdReference.set(jobClient.getJobID());
        jobClient = jobClient /*jobClient static public object in the main class*/;
    }@Override
    public void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable) {
        assert jobExecutionResult != null;
        jobExecutionResult.notify();
    }
});

使用代码：

Preconditions.checkNotNull(jobClient);
    final String status = jobClient.getJobStatus().get().name();
    if (status.equals(JobStatus.FAILED.name())) System.exit(1);