Flink Kafka Producer中的精确一次语义学

Question

一、 我正试图用Kafka信源和信宿测试Flink一次语义：

1. 运行flink应用程序，只需将消息从一个主题传输到另一个主题，并行度=1，检查点间隔20秒
2. 每2秒使用Python脚本生成具有递增整数的消息。
3. 使用read_committed隔离级别的控制台使用者读取输出主题。
4. 手动杀死TaskManager

我希望在输出主题中看到单调递增的整数，而不考虑TaskManager的终止和恢复。

但实际上，a在控制台使用者输出中看到了一些意想不到的东西：

32
33
34
35
36
37
38
39
40
-- TaskManagerKilled
32
34
35
36
40
41
46
31
33
37
38
39
42
43
44
45

看起来像是在输出主题中重播的检查点之间的所有消息。这应该是正确的行为还是我做错了什么？

恢复了一个快照：Flink UI

我的Flink代码：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getConfig().setAutoWatermarkInterval(1000);
        env.enableCheckpointing(20000, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data"));

        Properties producerProperty = new Properties();
        producerProperty.setProperty("bootstrap.servers", ...);
        producerProperty.setProperty("zookeeper.connect", ...);
        producerProperty.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,"10000");
        producerProperty.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"my-transaction");
        producerProperty.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        Properties consumerProperty = new Properties();
        consumerProperty.setProperty("bootstrap.servers", ...);
        consumerProperty.setProperty("zookeeper.connect", ...);
        consumerProperty.setProperty("group.id", "test2");

        FlinkKafkaConsumer<String> consumer1 = new FlinkKafkaConsumer<String>("stringTopic1", new ComplexStringSchema(), consumerProperty);
        consumer1.assignTimestampsAndWatermarks(new PeriodicAssigner());

        FlinkKafkaProducer<String> producer1 = new FlinkKafkaProducer<String>("test",  new KeyedSerializationSchemaWrapper(new SimpleStringSchema()), producerProperty, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        producer1.ignoreFailuresAfterTransactionTimeout();
        DataStreamSource<String> s1 = env.addSource(consumer1);
        s1.addSink(producer1);
        env.execute("Test");
    }

Answer 1

Flink以规则的、可配置的间隔生成检查点。当恢复检查点时，Flink会将状态回滚到输入流中最后一次检查点的位置（不一定与上次处理/消费的位置相同）。有不同的方法来确保精确一次语义学。您可以使用支持精确一次语义学的生产者（接收器），请参阅：Flink接收器中的容错保证。

或者，您可以在消费者中支持精确一次语义学。假设唯一整数与多个工作者保持一致（并行性

>

假设当前检查点id为Ckpt N。将所有已处理的整数（大事件情况下已处理事件的指纹）存储在Ckpt N状态。您可以通过让消费者实现ListCheckped接口来实现这一点，以将状态（指纹，或您情况下的整数）存储在Ckpt N中。

一旦Flink移动到下一个检查点（Ckpt N 1），过滤掉存储在Ckpt N状态中的所有整数，以确保只进行一次处理。以Ckpt N 1的状态（即丢弃Ckpt N的状态）存储未过滤的已处理整数（或已处理事件的指纹）。

您只需要存储发生在两个检查点之间的已处理事件（或您的情况下的整数）的指纹，然后在持久化新检查点时丢弃。

Answer 2

除了为精确一次语义学设置生产者之外，您还需要将消费者配置为仅读取来自kafka的已提交消息。默认情况下，消费者将读取已提交和未提交的消息。将此设置添加到您的消费者应该会让您更接近您想要的行为。

consumerProperties.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");