我正在尝试使用同步pullapi使用googlepubsub消息。这在apache beam google pubio连接器库中提供。我想用kafkaio把消耗的信息写给kafka。我想使用flinkrunner来执行作业，因为我们在gcp之外运行这个应用程序。
我面临的问题是，在gcp pubsub中消耗的消息没有得到确认。我已经确认本地kafka示例有从gcp pubsub消费的消息。gcpdataflow中的文档表明，当管道以数据接收器（在我的例子中是kafka）终止时，数据包就完成了。
但是，由于代码是在apache flink中运行的，而不是gcp数据流，因此我认为与确认提交的消息相关的某种回调不会被触发。
我做错什么了？

pipeline
                    .apply("Read  GCP PubSub Messages", PubsubIO.readStrings()
                            .fromSubscription(subscription)
                    )
                    .apply(ParseJsons.of(User.class))
                    .setCoder(SerializableCoder.of(User.class))
                    .apply("Filter-1", ParDo.of(new FilterTextFn()))
                    .apply(AsJsons.of(User.class).withMapper(new ObjectMapper()))
                    .apply("Write to Local Kafka",
                            KafkaIO.<Void,String>write()
                                    .withBootstrapServers("127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094")
                                    .withTopic("test-topic")
                                    .withValueSerializer((StringSerializer.class))
                                    .values()
                    );