Apache Beam PubSub Reader 异常

Question

我是运行一个带有 PubSub 源的管道，我遇到了一些奇怪的异常，因为我的管道崩溃了。我可以很好地处理一些元素 (3-10)，然后突然抛出以下两个错误消息之一。两者都没有告诉我我可能做错了什么，所以我删除了所有的转换，只留下了源代码，问题仍然存在。我只是将一些测试字符串发布到 PubSub。感谢任何帮助。

例外情况 1：

[WARNING]
java.lang.reflect.InvocationTargetException
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.codehaus.mojo.exec.ExecJavaMojo.run(ExecJavaMojo.java:293)
        at java.lang.Thread.run(Thread.java:724)
Caused by: java.lang.NullPointerException
        at org.apache.beam.sdk.io.PubsubUnboundedSource$PubsubReader.ackBatch(PubsubUnboundedSource.java:640)
        at org.apache.beam.sdk.io.PubsubUnboundedSource$PubsubCheckpoint.finalizeCheckpoint(PubsubUnboundedSource.java:313)
        at org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.getReader(UnboundedReadEvaluatorFactory.java:174)
        at org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.processElement(UnboundedReadEvaluatorFactory.java:127)
        at org.apache.beam.runners.direct.TransformExecutor.processElements(TransformExecutor.java:139)
        at org.apache.beam.runners.direct.TransformExecutor.run(TransformExecutor.java:107)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)

异常 2：

[WARNING]
java.lang.reflect.InvocationTargetException
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.codehaus.mojo.exec.ExecJavaMojo.run(ExecJavaMojo.java:293)
        at java.lang.Thread.run(Thread.java:724)
Caused by: java.lang.IllegalStateException: Cannot finalize a restored checkpoint
        at org.apache.beam.sdk.repackaged.com.google.common.base.Preconditions.checkState(Preconditions.java:444)
        at org.apache.beam.sdk.io.PubsubUnboundedSource$PubsubCheckpoint.finalizeCheckpoint(PubsubUnboundedSource.java:293)
        at org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.finishRead(UnboundedReadEvaluatorFactory.java:205)
        at org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.processElement(UnboundedReadEvaluatorFactory.java:142)
        at org.apache.beam.runners.direct.TransformExecutor.processElements(TransformExecutor.java:139)
        at org.apache.beam.runners.direct.TransformExecutor.run(TransformExecutor.java:107)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)

基本代码：

PipelineOptions options = PipelineOptionsFactory.create(); 
PubsubOptions dataflowOptions = options.as(PubsubOptions.class);
dataflowOptions.setStreaming(true);

Pipeline p = Pipeline.create(options);

p.apply(PubsubIO.<String>read().subscription("my-subscription")
    .withCoder(StringUtf8Coder.of())));

执行：

mvn compile exec:java -Dexec.mainClass=my.package.SalesTransactions -Dexec.args="--runner BlockingDataflowRunner --project=my-project --tempLocation=gs://my-project/tmp"

Answer 1

存在此问题是因为 DirectRunner 中的错误 (BEAM-1656) 和 PubsubCheckpoint 中的先决条件。

答案包含有关错误及其解决方法的更多信息。谢谢！

Apache Beam PubSub Reader 异常

Apache Beam PubSub Reader Exceptions

google-cloud-dataflow

google-cloud-pubsub

apache-beam