flink sink kafka的变乱提交征象料想

征象

查察flink源码时 sink kafka有变乱提交机制，查察源码发现是使用两阶段提交战略，而变乱提交是checkpoint完成后才实行，那么假如checkpoint设置隔断时间比力长时，变乱未提交之前，后端应该斲丧不到数据，而观察现实征象为写入kafka的斲丧数据可以立马斲丧。
测试用例

测试流程

  

• 编写任务1，设置较长的checkpoint时间，而且指定 CheckpointingMode.EXACTLY_ONCE，输出输出到kafka。  
  
• 编写任务2斲丧任务的结果topic，打印控制台，验证结果。  
  
• 根据征象查察源码，分析缘故起因。
  

测试用例

测试任务1

  

1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
2.         env.setParallelism(1);
4.         env.enableCheckpointing(1000*60l, CheckpointingMode.EXACTLY_ONCE);
5.         env.getCheckpointConfig().setCheckpointStorage("file:///flink/checkpoint");
7. //        超时时间，checkpoint没在时间内完成则丢弃
8.         env.getCheckpointConfig().setCheckpointTimeout(50000L); //10秒
9.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
10.         env.getCheckpointConfig().setTolerableCheckpointFailureNumber(1);
11.         //最小间隔时间（前一次结束时间，与下一次开始时间间隔）
12.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
13. //        当 Flink 任务取消时，保留外部保存的 checkpoint 信息
15.         
16.         KafkaSource<String> source = KafkaSource.<String>builder()
17.                 .setBootstrapServers("127.0.0.1:9092")
18.                 .setTopics("test001")
19.                 .setGroupId("my-group")
20. //                .setStartingOffsets(OffsetsInitializer())
21.                 .setStartingOffsets(OffsetsInitializer.committedOffsets())
22.                 .setValueOnlyDeserializer(new SimpleStringSchema())
23.                 .build();
26.         DataStreamSource<String> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
27.         // 从文件读取数据
28. //        DataStream<SensorReading> dataStream = env.addSource( new SourceTest4.MySensorSource() );
29.         DataStream<String> map = kafkaSource.map(new MapFunction<String, String>() {
30.             @Override
31.             public String map(String s) throws Exception {
32.             
33.                 return s;
34.             }
35.         });
36.         Properties properties = new Properties();
37. // 根据上面的介绍自己计算这边的超时时间，满足条件即可
38.         properties.setProperty("transaction.timeout.ms","900000");
39. //        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
40.         KafkaSink<String> sink = KafkaSink.<String>builder()
41.                 .setBootstrapServers("192.168.65.128:9092")
42.                 .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
43.                         .setTopic("test002")
44.                         .setValueSerializationSchema(new SimpleStringSchema())
45.                         .build()
46.                 )
47.                 .setKafkaProducerConfig(properties)
48.                 .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
49. //                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
50.                 .setTransactionalIdPrefix("flink-xhaodream-")
51.                 .build();
52.         map.sinkTo(sink);
53.         // 打印输出
54.         env.execute();

复制代码

测试任务2

  

1.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
2.         env.setParallelism(1);
4.         env.enableCheckpointing(1000*150l, CheckpointingMode.EXACTLY_ONCE);
5.         env.getCheckpointConfig().setCheckpointStorage("file:///flink/checkpoint");
6. //        当 Flink 任务取消时，保留外部保存的 checkpoint 信息
7.         Properties properties1 = new Properties();
8. //        properties1.put("isolation.level","read_committed");
9.         KafkaSource<String> source = KafkaSource.<String>builder()
10.                 .setBootstrapServers("127.0.0.1:9092")
11.                 .setTopics("test002")
12.                 .setGroupId("my-group2")
13.                 .setProperties(properties1)
14.                 .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
15.                 .setValueOnlyDeserializer(new SimpleStringSchema())
16.                 .build();
17.         DataStreamSource<String> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
18.         kafkaSource.print(" test2接受数据");
19.         // 打印输出
20.         env.execute();

复制代码

测试结果分析

测试结果：

任务1开启后，无论是否实行checkpoint，任务checkpoint都可以正常斲丧数据，与预期不符合。
缘故起因排查

查察kafkaSink 的源码，找到跟与两阶段提交干系的代码，1.18源码中TwoPhaseCommittingSink有重构。kafkasink实现TwoPhaseCommittingSink接口实现，创建Commiter和Writer。
  

1. @PublicEvolving
2. public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
3.     PrecommittingSinkWriter<InputT, CommT> createWriter(Sink.InitContext var1) throws IOException;
5.     Committer<CommT> createCommitter() throws IOException;
7.     SimpleVersi

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。