526互联

【flink】重启kafka消费者

发布时间 2023-04-23 10:51:20作者: xiaoyu_jane
public class KafkaConsumerJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test-group");

      //创建了一个Flink Kafka消费者，并在StreamExecutionEnvironment中添加它作为数据源
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("myTopic", new SimpleStringSchema(), props);
        DataStream<String> stream = env.addSource(kafkaConsumer);

        stream.print();
        env.execute("Kafka Consumer Job");
    }

//在Flink集群上运行该应用程序，则需要考虑应用程序在多个TaskManager上运行的情况，并确保在所有TaskManager上都执行相同的重启逻辑
    public void restart() {
        // 停止现有的Kafka消费者或生产者，关闭与Kafka的连接。
        // 销毁现有的Kafka消费者或生产者实例。
        kafkaConsumer.close();
        // 创建一个新的Kafka消费者或生产者实例，并重新连接到Kafka集群。
        kafkaConsumer = new FlinkKafkaConsumer<>("myTopic", new SimpleStringSchema(), props);
        stream = env.addSource(kafkaConsumer);
    }
}