Apache Flink提供了用于管理MQTT消费者偏移量的API。下面是一个简单的示例。

首先，需要引入相关依赖：

    org.apache.flink
    flink-connector-mqtt_2.12
    ${flink.version}

然后，可以使用以下代码创建一个带有偏移量管理的MQTT连接：

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "my-group-id");
properties.setProperty("enable.auto.commit", "false");

MqttSource source = MqttSource.builder()
        .setUrls(Collections.singletonList("tcp://localhost:1883"))
        .setTopic("my-topic")
        .setClientId("my-client-id")
        .setDeserializationSchema(new SimpleStringSchema())
        .setProperties(properties)
        .setStartingOffsets(OffsetsInitializer.latest())
        .setCleanupMode(CleanupMode.COMPACT)
        .build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(source)
        .map(value -> {
            // do something
            return value;
        })
        .addSink(/* your sink */);

env.execute("My Job");

在上述代码中，我们可以看到以下主要操作：

1. 使用MqttSource对象创建一个带有偏移量管理的MQTT连接。
2. 使用setStartingOffsets方法设置消费起始偏移量。
3. 在 MqttSource 中使用CleanupMode 枚举类型来设置清理策略。

通过以上操作，我们可以轻松地为Apache Flink中的MQTT消费者添加偏移量管理的功能。