如何获取作业提交时间并在Flink应用程序中使用？

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class JobSubmissionTimeExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置时间特性为EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 创建数据源，每个元素包含一个时间戳和一个值
        DataStream<Tuple2<Long, String>> input = env.fromElements(
                Tuple2.of(System.currentTimeMillis(), "value1"),
                Tuple2.of(System.currentTimeMillis(), "value2"),
                Tuple2.of(System.currentTimeMillis(), "value3")
        );

        // 提取时间戳，并分配水位线
        DataStream<Tuple2<Long, String>> withTimestampsAndWatermarks = input
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<Long, String>>(Time.seconds(10)) {
                    @Override
                    public long extractTimestamp(Tuple2<Long, String> element) {
                        return element.f0;
                    }
                });

        // 使用ProcessFunction获取作业提交时间并输出
        withTimestampsAndWatermarks
                .process(new ProcessFunction<Tuple2<Long, String>, String>() {
                    @Override
                    public void processElement(Tuple2<Long, String> value, Context ctx, Collector<String> out) throws Exception {
                        long jobSubmissionTime = ctx.timestamp();
                        out.collect("Job submission time: " + jobSubmissionTime);
                    }
                })
                .print();

        env.execute("Job Submission Time Example");
    }
}