flink保存消费kafka的偏移量

在 Flink 中，可以通过 `KafkaConsumer` 提供的 `FlinkKafkaConsumer` 来消费 Kafka 中的数据。而保存消费 Kafka 的偏移量可以使用 Flink 提供的 `FlinkKafkaConsumer` 的 `enableCheckpointing` 方法来实现。

具体做法如下：

1. 在创建 `FlinkKafkaConsumer` 对象时，设置 `enableCheckpointing` 方法开启 Flink 的 checkpoint 机制。

FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties);
kafkaConsumer.setStartFromEarliest();
kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
env.enableCheckpointing(5000);

2. 设置 `setCommitOffsetsOnCheckpoints` 方法为 `true`，表示 Flink 在进行 checkpoint 时会自动提交当前消费 Kafka 的偏移量。

3. 在 Flink 应用程序中，可以通过实现 `CheckpointedFunction` 接口来手动管理 Kafka 偏移量的保存和恢复。

public class KafkaConsumer implements SourceFunction<String>, CheckpointedFunction {
    private transient ListState<Long> offsetState;
    private volatile boolean isRunning = true;
    private long offset = 0L;

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(offset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> offsetStateDescriptor = new ListStateDescriptor<>("offsets", Long.class);
        offsetState = context.getOperatorStateStore().getListState(offsetStateDescriptor);

        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                this.offset = offset;
            }
        }
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                offset = record.offset();
                ctx.collect(record.value());
            }
        }
    }
}

在 `snapshotState` 方法中，将当前消费 Kafka 的偏移量保存到 `offsetState` 状态变量中。在 `initializeState` 方法中，可以从状态变量中恢复保存的偏移量。在 `run` 方法中，每次消费 Kafka 中的数据时，都会更新 `offset` 变量为当前消费的偏移量。

这样，当 Flink 应用程序出现故障，重新启动时，就可以从保存的偏移量处继续消费 Kafka 中的数据，避免了数据的重复消费。