Apache Beam：PubsubReader在NPE时失败

Question

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:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Answer 1

此问题的存在是因为DirectRunner中有一个Bug(BEAM-1656)和PubSubcheckpoint中有一个前置条件。DirectRunner中的bug在pull request 2237中得到了修复，该bug被合并到Github主分支中，但在0.6.0版本之后。

在使用DirectRunner时，更新到0.7.0或从github HEAD构建可以解决这个问题。

要更新到当前的夜间构建，您必须将以下存储库添加到项目的pom.xml中。包含此修复的beam-runners-direct-java模块的最早版本是0.7.0-20170316.070901-9，但并非所有模块都是用此特定版本构建的，因此您可能必须指定单独兼容的版本或使用0.7.0-snapshot

    <repositories>
      <repository>
        <id>apache.snapshots</id>
        <name>Apache Development Snapshot Repository</name>

 <url>https://repository.apache.org/content/repositories/snapshots/</url>
        <releases>
          <enabled>false</enabled>
        </releases>
        <snapshots>
          <enabled>true</enabled>
        </snapshots>
      </repository>

    </repositories>