flink 多算子怎么提交offset

在 Flink 中，提交 Offset 的时机取决于使用的消费者类型。对于 Flink Kafka 消费者，在多个算子中处理同一个 Kafka 主题时，需要确保所有算子都使用相同的 Group ID，并且只有最后一个算子才提交 Offset。这是因为在 Flink Kafka 消费者中，Offset 是按消费者组提交的，因此如果多个算子使用不同的 Group ID 提交 Offset，将会发生重复消费的情况。

对于 Flink 自定义消费者，在多个算子中处理同一个数据流时，可以在每个算子中手动提交 Offset。通常情况下，在每个算子中使用 Flink 的 `map` 或 `flatmap` 函数时，可以使用 `RichMapFunction` 或 `RichFlatMapFunction` 类来覆盖 `open` 函数，并在其中创建 Kafka Consumer 实例和 Kafka Consumer Offset 操作对象。在处理完每个记录后，可以调用 `commitSync` 或 `commitAsync` 函数来手动提交 Offset。需要注意的是，如果在处理记录时发生异常，需要使用 `rollback` 函数来回滚 Offset。

以下是使用 Flink 自定义消费者提交 Offset 的示例代码：

DataStreamSource<String> stream = env.addSource(new MySource());

stream
    .map(new MyMapFunction())
    .map(new AnotherMapFunction())
    .addSink(new MySink());

public static class MyMapFunction extends RichMapFunction<String, String> {
    private transient KafkaConsumer<String, String> consumer;
    private transient OffsetCommitCallback offsetCommitCallback;

    @Override
    public void open(Configuration parameters) throws Exception {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "my-group-id");

        consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
        consumer.subscribe(Collections.singletonList("my-topic"));

        offsetCommitCallback = new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    // 处理提交 Offset 异常
                }
            }
        };
    }

    @Override
    public String map(String value) throws Exception {
        // 处理记录
        consumer.commitAsync(offsetCommitCallback);
        return value;
    }

    @Override
    public void close() throws Exception {
        consumer.close();
    }
}