Flink:管道如何优雅地关闭自身？

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

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

        // 数据源
        env.addSource(new SourceFunction<Integer>() {
            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                while (isRunning) {
                    ctx.collect(1);
                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        })
        .addSink(new SinkFunction<Integer>() {
            @Override
            public void invoke(Integer value, Context context) throws Exception {
                System.out.println("Received: " + value);
            }
        });

        // 注册关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            env.close();
            System.out.println("Flink job gracefully shut down.");
        }));

        env.execute("Graceful Shutdown Example");
    }
}