Flink检查点不重放保存点/检查点期间正在处理的kafka事件

Question

我想在flink中测试一次端到端的处理。我的工作是：

Kafka资料来源-

我在mapper1中放了一个< code > thread . sleep(100000)，然后运行了这个作业。我在停止作业时获取了保存点，然后从mapper1中删除了< code > thread . sleep(100000)，我希望该事件应该会被重放，因为它没有下沉。但这并没有发生，乔布斯正在等待新的事件。

我的Kafka资料来源：

KafkaSource.<String>builder()
                .setBootstrapServers(consumerConfig.getBrokers())
                .setTopics(consumerConfig.getTopic())
                .setGroupId(consumerConfig.getGroupId())
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setProperty("commit.offsets.on.checkpoint", "true")
                .build();

我的Kafka水槽:

KafkaSink.<String>builder()
                .setBootstrapServers(producerConfig.getBootstrapServers())
                .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(producerConfig.getTopic())
                        .setValueSerializationSchema(new SimpleStringSchema()).build())
                .build();

我的flink作业环境设置：

StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.enableCheckpointing(2000);
        environment.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        environment.getCheckpointConfig().setMinPauseBetweenCheckpoints(100);
        environment.getCheckpointConfig().setCheckpointTimeout(60000);
        environment.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        environment.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        environment.getCheckpointConfig().setCheckpointTimeout(1000);
        environment.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        environment.getCheckpointConfig().enableUnalignedCheckpoints();
        environment.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        Configuration configuration = new Configuration();
        configuration.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
        environment.configure(configuration);

我到底做错了什么？我希望在取消/停止作业期间正在进行的任何事件都应该重新开始。

编辑1:我注意到我的kafka对我的flink的kafka-source消费者群体显示出了抵消滞后。我假设这意味着我的检查点运行正常，对吗？

我还观察到，当我从检查点重新启动作业时，它并未开始从剩余的偏移量中消耗，而我将使用者偏移量设置为FIRST。我必须发送更多事件来触发kafka源端的消耗，然后它消耗了所有事件。

Answer 1

对于“恰好一次”，您必须在针对同一Kafka集群运行的所有应用程序之间提供唯一的< code > TransactionalIdPrefix (与传统的< code>FlinkKafkaConsumer相比，这是一个变化):

KafkaSink<T> sink =
        KafkaSink.<T>builder()
                .setBootstrapServers(...)
                .setKafkaProducerConfig(...)
                .setRecordSerializer(...)
                .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix("unique-id-for-your-app")
                .build();

当从检查点恢复时，Flink总是使用存储在检查点中的偏移量，而不是在代码中配置或存储在代理中的偏移量。