FlinkKafkaConsumer如何停止拉取消息

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

public class KafkaConsumerExample {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置Kafka消费者
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-consumer-group");

        // 创建FlinkKafkaConsumer实例
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), properties);

        // 添加Kafka消费者到执行环境
        env.addSource(kafkaConsumer).print();

        // 启动消息拉取和消费
        env.execute("Kafka Consumer Example");

        // 停止消息拉取
        kafkaConsumer.cancel();
    }
}