Flink任务在由不可预测异常导致任务挂掉,然后自动重新提交之后会重复消费kafka里的数据怎么解决?

如代码所示,同一个flink任务里有多个处理不同数据流的flatMap,从kafka不同的topic中取数据进行处理,其中stream1有不可预测的异常没有抓住,或者因为环境问题导致整个任务被cancel,然后自动重新启动,重启任务之后都会概率性出现有数条kafka里的数据被重复处理,有木有大神知道怎么解决啊?

StreamExecutionEnvitoment env = StreamExecutionEnvitoment.getExecutionEnvitoment();
env.enableCheckpointing(5000);

FlinkKafkaConsumer011<String> myConsumer1 = new FlinkKafkaConsumer011<>(topic1, Charset.forName("ISO8859-1"), prop);
myConsumer1.setStartFromLatest();
DataStream<String> stream1 = env.addSource(myConsumer).name(topic);
stream1.flatMap(new Handle1()).setParallelism(1).setMaxParallelism(1).addSink(new Sink1());

FlinkKafkaConsumer011<String> myConsumer2 = new FlinkKafkaConsumer011<>(topic2, Charset.forName("ISO8859-1"), prop);
myConsumer2.setStartFromLatest();
DataStream<String> stream2 = env.addSource(myConsumer).name(topic);
stream2.flatMap(new Handle2()).setParallelism(1).setMaxParallelism(1).addSink(new Sink2());

1个回答

flink支持失败点恢复机制

Csdn user default icon
上传中...
上传图片
插入图片
抄袭、复制答案,以达到刷声望分或其他目的的行为,在CSDN问答是严格禁止的,一经发现立刻封号。是时候展现真正的技术了!
立即提问
相关内容推荐