无法在Flink新Kafka消费者api中的检查点向Kafka提交消费补偿（1.14）

Question

我指的是Kafka源代码连接器的Flink 1.14版本，代码如下。

我期待以下要求。

• 在最新的应用程序开始时，必须阅读Kafka主题的最新偏移量
• 在检查点上，它必须将消耗的偏移量提交给Kafka
• 重新启动后（当应用程序手动终止/系统错误时），它必须从最后提交的偏移量中选取，并且必须消耗使用者延迟，然后再使用新的事件提要

有了Flink新的KafkaConsumer API（KafkaSource），我面临以下问题

• 能够完成上述要求，但不能在检查点（500ms）上提交消耗的偏移量。它宁愿在2s或3s后提交。

在2/3秒内手动关闭应用程序并重新启动时。由于上次使用的消息未提交，因此会读取两次（重复）。

为了交叉检查这个功能，我尝试了Flink Kafka的旧消费者应用编程接口（Flink Kafka消费者）。在那里，它是完美的工作。当消息被立即消费时，它被提交回Kafka。

接下来的步骤

• 建立Kafka环境

如果我丢失了任何东西或需要添加任何财产，请提出建议。

 @Test
    public void test() throws Exception {

        System.out.println("FlinkKafkaStreamsTest started ..");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.enableCheckpointing(500);
        env.setParallelism(4);

        Properties propertiesOld = new Properties();
        Properties properties = new Properties();
        String inputTopic = "input_topic";
        String bootStrapServers = "localhost:29092";
        String groupId_older = "older_test1";
        String groupId = "test1";

        propertiesOld.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        propertiesOld.put(ConsumerConfig.GROUP_ID_CONFIG, groupId_older);
        propertiesOld.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);


        /******************** Old Kafka API **************/
        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(inputTopic,
                new KRecordDes(),
                propertiesOld);
        flinkKafkaConsumer.setStartFromGroupOffsets();
        env.addSource(flinkKafkaConsumer).print("old-api");


        /******************** New Kafka API **************/
        KafkaSourceBuilder<String> sourceBuilder = KafkaSource.<String>builder()
                .setBootstrapServers(bootStrapServers)
                .setTopics(inputTopic)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setProperty("enable.auto.commit", "false")
                .setProperty("commit.offsets.on.checkpoint", "true")
                .setProperties(properties)
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST));

        KafkaSource<String> kafkaSource = sourceBuilder.build();

        SingleOutputStreamOperator<String> source = env
                .fromSource(kafkaSource, WatermarkStrategy.forMonotonousTimestamps(), "Kafka Source");

        source.print("new-api");

        env.execute();
    }
    static class KRecordDes implements  KafkaDeserializationSchema<String>{
        @Override
        public TypeInformation<String> getProducedType() {
            return TypeInformation.of(String.class);
        }
        @Override
        public boolean isEndOfStream(String nextElement) {
            return false;
        }
        @Override
        public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
            return new String(consumerRecord.value());
        }
    }

注意：我还有其他要求，我希望Flink Kafka有界源代码读取器在相同的代码中，这在新的API（KafkaSource）中可用。

Answer 1

Kafka的文件来源：

请注意，Kafka source不依赖提交的偏移量来实现容错。提交抵销仅用于公开消费者和消费群体的进度以进行监控。

当Flink作业从失败中恢复时，它将从最近成功的检查点恢复状态，并从存储在该检查点中的偏移恢复消耗，因此检查点之后的记录将被“重播”一点。由于您使用的是打印接收器，它不支持完全一次语义，因此您将看到重复的记录，这些记录实际上是最近一次成功检查点之后的记录。

对于您提到的偏移量提交的2-3秒延迟，这是因为实现了SourceReaderBase。简而言之，SplitFetcher管理一个任务队列，当一个offset commit任务被推到队列中时，它不会被执行，直到调用KafkaConsumer#poll（）的正在运行的fetch任务超时。如果交通量很小，延误可能会更长。但请注意，这不会影响正确性：KafkaSource不使用提交的偏移量进行容错。