与InProcessPipelineRunner执行时，PubsubReader失败，并出现NullPointerException [英] PubsubReader failed with NullPointerException when executed with InProcessPipelineRunner

问题描述

我有一个简单的管道，仅执行读取PubsubIO.Read.subscription.消耗约200个元素后，每次运行都会失败，但以下情况除外:

I have simple pipeline that just performs the reading PubsubIO.Read.subscription. It fails for every run after consuming about 200 elements with the following exception:

[error] (run-main-0) java.lang.RuntimeException: java.lang.NullPointerException
java.lang.RuntimeException: java.lang.NullPointerException
    at  com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:281)
    at com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:69)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at com.sandbox.WriteLogsToBQ.main(WriteLogsToBQ.java:296)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
Caused by: java.lang.NullPointerException
    at com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource$PubsubReader.ackBatch(PubsubUnboundedSource.java:612)
    at com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource$PubsubCheckpoint.finalizeCheckpoint(PubsubUnboundedSource.java:297)
    at com.google.cloud.dataflow.sdk.runners.inprocess.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.startReader(UnboundedReadEvaluatorFactory.java:203)
    at com.google.cloud.dataflow.sdk.runners.inprocess.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.finishBundle(UnboundedReadEvaluatorFactory.java:172)
    at com.google.cloud.dataflow.sdk.runners.inprocess.TransformExecutor.finishBundle(TransformExecutor.java:163)
    at com.google.cloud.dataflow.sdk.runners.inprocess.TransformExecutor.run(TransformExecutor.java:119)
    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)

我正在使用SDK版本1.9.0. 对于1.6.1(成功获取> 10k元素)而言，这并未发生. 有人知道解决方法吗?我还注意到1.9.0的获取速度比1.6.1快得多.对于1.6.1，似乎使用了10个元素的批处理.

I am using SDK version 1.9.0. This has not happened with 1.6.1 (successfully fetched >10k elements). Does anyone know workaround? I also noticed that 1.9.0 fetching much faster than 1.6.1. For 1.6.1 it seems to use batches of 10 elements.

经过阅读表格主题的测试:

Tested with reading form topic:

p.apply(PubsubIO.Read.named("reading_topic_test").topic("projects/***/topics/SL_LogLogin"))

订阅是自动生成的，但随后管道失败，并显示以下信息:

Subscription was automatically generated, but then pipeline failed with:

Jan 05, 2017 2:06:33 PM com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource apply
WARNING: Created subscription null to topic NestedValueProvider{value=NestedValueProvider{value=StaticValueProvider{value=projects/***/topics/SL_LogLogin}}}. 
Note this subscription WILL NOT be deleted when the pipeline terminates
[error] (run-main-0) com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder$PopulateDisplayDataException: 
Error while populating display data for component:com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource$StatsFn
com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder$PopulateDisplayDataException: 
Error while populating display data for component: com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource$StatsFn
    at    com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:664)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:643)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:637)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.populateDisplayData(ParDo.java:1266)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.access$200(ParDo.java:457)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$Bound.populateDisplayData(ParDo.java:816)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:657)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:643)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:637)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.forRoot(DisplayData.java:630)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.access$000(DisplayData.java:617)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData.from(DisplayData.java:76)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.evaluateDisplayData(DisplayDataValidator.java:47)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.access$100(DisplayDataValidator.java:29)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator$Visitor.visitTransform(DisplayDataValidator.java:62)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:221)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:103)
    at com.google.cloud.dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:260)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.validateTransforms(DisplayDataValidator.java:43)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.validatePipeline(DisplayDataValidator.java:35)
    at com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:245)
    at com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:69)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at com.sandbox.WriteLogsToBQ.main(WriteLogsToBQ.java:303)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
Caused by: java.lang.NullPointerException: Input display value cannot be null
    at com.google.cloud.dataflow.sdk.repackaged.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:228)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.addItemIf(DisplayData.java:707)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.add(DisplayData.java:685)
    at com.google.cloud.dataflow.sdk.io.PubsubUnboundedSource$StatsFn.populateDisplayData(PubsubUnboundedSource.java:1147)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:657)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:643)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:637)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.populateDisplayData(ParDo.java:1266)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.access$200(ParDo.java:457)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$Bound.populateDisplayData(ParDo.java:816)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:657)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:643)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.include(DisplayData.java:637)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.forRoot(DisplayData.java:630)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData$InternalBuilder.access$000(DisplayData.java:617)
    at com.google.cloud.dataflow.sdk.transforms.display.DisplayData.from(DisplayData.java:76)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.evaluateDisplayData(DisplayDataValidator.java:47)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.access$100(DisplayDataValidator.java:29)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator$Visitor.visitTransform(DisplayDataValidator.java:62)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:221)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:103)
    at com.google.cloud.dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:260)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.validateTransforms(DisplayDataValidator.java:43)
    at com.google.cloud.dataflow.sdk.runners.inprocess.DisplayDataValidator.validatePipeline(DisplayDataValidator.java:35)
    at com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:245)
    at com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner.run(InProcessPipelineRunner.java:69)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at com.sandbox.WriteLogsToBQ.main(WriteLogsToBQ.java:303)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)

推荐答案

https处理://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/547 .抱歉造成麻烦.

根本原因是孵化器光束与GoogleDataflowSDK之间发生向后移植错误，该错误无法初始化在按作业订阅的情况下返回的订阅.

The root cause was a backporting error between incubator-beam and GoogleDataflowSDK, which failed to initialize the subscription that was being returned in the case of a per-job subscription.

这篇关于与InProcessPipelineRunner执行时，PubsubReader失败，并出现NullPointerException的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！