如何从FlinkKafkaConsumer获取最新的消息偏移量？

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;

import java.util.Map;

public class KafkaOffsetExample {

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

        // 设置Kafka连接信息和消费者组ID
        String kafkaServers = "localhost:9092";
        String kafkaTopic = "my-topic";
        String consumerGroup = "my-consumer-group";

        // 创建FlinkKafkaConsumer实例
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(kafkaTopic, new SimpleStringSchema(), properties);
        kafkaConsumer.setStartFromLatest();

        // 将FlinkKafkaConsumer添加到Flink作业中
        DataStream<String> stream = env.addSource(kafkaConsumer);

        // 为数据流分配时间戳和水印
        stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(String element) {
                // 从消息中提取时间戳
                return Long.parseLong(element.split(",")[0]);
            }
        });

        // 开启将消费者的偏移量保存到检查点的功能
        env.enableCheckpointing(5000);
        kafkaConsumer.setCommitOffsetsOnCheckpoints(true);

        // 获取运行时上下文
        Map<KafkaTopicPartition, Long> offsets = getOffsetsFromKafkaConsumer(kafkaConsumer.getRuntimeContext());

        // 遍历每个分区，获取最新消息偏移量
        for (Map.Entry<KafkaTopicPartition, Long> entry : offsets.entrySet()) {
            KafkaTopicPartition partition = entry.getKey();
            long offset = entry.getValue();
            System.out.println("Partition: " + partition + ", Offset: " + offset);
        }

        env.execute("Kafka Offset Example");
    }

    private static Map<KafkaTopicPartition, Long> getOffsetsFromKafkaConsumer(RuntimeContext context) {
        FlinkKafkaConsumer<String> kafkaConsumer = (FlinkKafkaConsumer<String>) context.getKafkaConsumer();
        return kafkaConsumer.assignment()
                .stream()
                .collect(Collectors.toMap(partition -> partition, kafkaConsumer::position));
    }
}