我有一个Beam管道来消费流事件，它有多个阶段（PTransforms）来处理它们。

pipeline.apply("Read Data from Stream", StreamReader.read())
            .apply("Decode event and extract relevant fields", ParDo.of(new DecodeExtractFields()))
            .apply("Deduplicate process", ParDo.of(new Deduplication()))
            .apply("Conversion, Mapping and Persisting", ParDo.of(new DataTransformer()))
            .apply("Build Kafka Message", ParDo.of(new PrepareMessage()))
            .apply("Publish", ParDo.of(new PublishMessage()))
            .apply("Commit offset", ParDo.of(new CommitOffset()));

使用KafkaIO和StreamReader.read()方法实现读取的流事件是这样的：

public static KafkaIO.Read<String, String> read() {
        return KafkaIO.<String, String>read()
                .withBootstrapServers(Constants.BOOTSTRAP_SERVER)
                .withTopics(Constants.KAFKA_TOPICS)
                .withConsumerConfigUpdates(Constants.CONSUMER_PROPERTIES)
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(StringDeserializer.class);
    }

在我们通过KafkaIO读取流事件/消息后，我们可以提交偏移量。我需要做的是手动提交偏移量，在执行所有先前PTransform时的最后一个Commit offset PTransform内。
原因是，我正在管道中间执行一些转换、Map和持久化，当所有事情都顺利完成时，我需要提交偏移量。通过这样做，如果处理在中间失败，我可以再次使用相同的记录/事件并进行处理。
我的问题是，我如何手动提交偏移？如果可以共享资源/示例代码，请表示感谢。