java如何在弗林克设立卡夫卡委员会？

Question

我已经通过以下方式设置了承诺抵销：

properties.setProperty("auto.commit.interval.ms", "1000");
properties.setProperty("auto.commit.enable", "true");
properties.setProperty("enable.auto.commit", "true");

FlinkKafkaConsumer08<MobilePageEvent> kafkaConsumer =
            new FlinkKafkaConsumer08<>(
                    "mobile-event.page-resource", SCHEMA, properties);

但在web UI中，我得到了无效的提交偏移量：

Answer 1

启用检查点后，它现在可以工作了：

StreamExecutionEnvironment environment =
            StreamExecutionEnvironment.getExecutionEnvironment();

environment.enableCheckpointing(5000);