确认 Apache Beam 上的 Google Pub/Sub 消息

Question

我正在尝试使用以下代码从 pub/sub 读取数据

Read<String> pubsub = PubsubIO.<String>read().topic("projects/<projectId>/topics/<topic>").subscription("projects/<projectId>/subscriptions/<subscription>").withCoder(StringUtf8Coder.of()).withAttributes(new SimpleFunction<PubsubMessage,String>() {
    @Override
    public String apply(PubsubMessage input) {
        LOG.info("hola " + input.getAttributeMap());
        return new String(input.getMessage());
    }
});
PCollection<String> pps = p.apply(pubsub)
        .apply(
                Window.<String>into(
                    FixedWindows.of(Duration.standardSeconds(15))));
pps.apply("printdata",ParDo.of(new DoFn<String, String>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
        LOG.info("hola amigo "+c.element());
        c.output(c.element());
    }
  }));

与我在 NodeJS 上收到的消息相比，我收到的消息将包含在 data 字段中。我怎样才能得到 ackId 字段（我以后可以用它来确认消息）？我正在打印的属性图是 null。有没有其他方法可以确认所有消息而不必弄清楚 ackId？

Answer 1

PubsubIO reader 负责确认消息。它与跑步者的检查点行为有关。具体来说，只有当生成的元素已被检查点时，源才会确认消息。

在这种情况下，您应该查看 Flink 运行程序检查点何时显示该源的状态信息。我相信这与检查点频率的 Flink 配置有关。

确认 Apache Beam 上的 Google Pub/Sub 消息

Acknowledge Google Pub/Sub message on Apache Beam

java

google-cloud-pubsub

apache-beam

apache-beam-io