Как установить kafka CommitOffset в Flink?

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);

Но в веб-интерфейсе я получаю недопустимые фиксированные смещения:

Answer 1

После включения проверки наведения теперь работает:

StreamExecutionEnvironment environment =
            StreamExecutionEnvironment.getExecutionEnvironment();

environment.enableCheckpointing(5000);