Flink独立集群在处理作业状态/处理偏移量时是否支持优雅手动重启？

Flink独立集群在处理作业状态/处理偏移量时是支持优雅手动重启的。下面是一个包含代码示例的解决方法：

首先，你需要使用Checkpointing来启用Flink的检查点机制。检查点机制可以在作业运行过程中周期性地保存作业的状态和处理进度。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000); // 设置检查点间隔为5秒

// 设置检查点配置，例如保存到文件系统中
env.getCheckpointConfig().setCheckpointStorage("hdfs://localhost:9000/flink-checkpoints");

// 启动Flink作业
DataStream<String> dataStream = env.fromElements("message");
dataStream.addSink(new MySinkFunction());
env.execute("MyJob");

然后，在你的应用程序代码中，你可以使用CancellationException来触发优雅的手动重启。当你想要重启作业时，你可以通过调用cancel()方法来取消正在运行的作业。

// 在某个地方触发优雅的手动重启
JobID jobId = ...; // 你需要获取正在运行的作业的JobID

// 创建一个Flink客户端来与集群通信
Configuration config = new Configuration();
config.set(JobManagerOptions.ADDRESS, "localhost");
config.set(JobManagerOptions.PORT, 8081);
config.set(RestOptions.RETRY_MAX_ATTEMPTS, 1); // 设置最大重试次数为1，以便能够及时退出

RestClusterClient client = new RestClusterClient<>(config, "localhost:8081");

// 取消正在运行的作业
client.cancel(jobId);

// 等待作业完成
client.waitUntilJobFinished(jobId);

// 重新启动作业
client.submitJob(new MyJobGraph());

// 关闭客户端
client.close();

通过以上代码，你可以在需要时优雅地取消正在运行的作业，并重新启动它。这样，你就可以处理作业状态和处理偏移量，从上次检查点的位置继续处理数据。