flink报空指针

[九思]

 请教老师，flink on yarn，往rabbitmq写数据 或者 消费rabbitmq数据，都会报空指针，啥意思呢？本地idea都是正常的。flink包版本是1.8的，flink环境之前搭的1.7

2019-07-22 11:32:12.309 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Source: Custom Source (1/1) (85cfdb83f536b26e07ca2aa4a1b66302) switched from DEPLOYING

 to FAILED.

java.lang.NullPointerException: null

        at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1175)

        at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212)

        at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190)

        at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:51)

        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)

        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)

        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)

        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)

        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405)

        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689)

        at java.lang.Thread.run(Thread.java:745)

2019-07-22 11:32:12.310 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Freeing task resources for Source: Custom Source (1/1) (85cfdb83f536b26e07ca2aa4a1b663

02).

2019-07-22 11:32:12.321 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Ensuring all FileSystem streams are closed for task Source: Custom Source (1/1) (85cfd

b83f536b26e07ca2aa4a1b66302) [FAILED]

2019-07-22 11:32:12.331 [flink-akka.actor.default-dispatcher-3] INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor  - Un-registering task and sending final execution state FAILED to Job

Manager for task Source: Custom Source 85cfdb83f536b26e07ca2aa4a1b66302.

  

[zhisheng]

hi 九思：
    从提供的日志来看是看不出什么问题导致的空指针异常，有更详细的日志或者把你的代码提供出来吗？或者可以简单参考我自己之前写的博客（写入数据到
rabbitmq）：http://1t.click/uh6
希望对你有帮助！

九思 <[hidden email]> 于2019年7月22日周一 下午1:32写道：

>  请教老师，flink on yarn，往rabbitmq写数据 或者
> 消费rabbitmq数据，都会报空指针，啥意思呢？本地idea都是正常的。flink包版本是1.8的，flink环境之前搭的1.7
>
> 2019-07-22 11:32:12.309 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Source: Custom Source (1/1)
> (85cfdb83f536b26e07ca2aa4a1b66302) switched from DEPLOYING
>  to FAILED.
> java.lang.NullPointerException: null
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1175)
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212)
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190)
>         at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:51)
>         at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native
> Method)
>         at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
>         at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>         at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>         at
> org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405)
>         at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689)
>         at java.lang.Thread.run(Thread.java:745)
> 2019-07-22 11:32:12.310 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Freeing task resources for
> Source: Custom Source (1/1) (85cfdb83f536b26e07ca2aa4a1b663
> 02).
> 2019-07-22 11:32:12.321 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Ensuring all FileSystem
> streams are closed for task Source: Custom Source (1/1) (85cfd
> b83f536b26e07ca2aa4a1b66302) [FAILED]
> 2019-07-22 11:32:12.331 [flink-akka.actor.default-dispatcher-3] INFO
> org.apache.flink.runtime.taskexecutor.TaskExecutor  - Un-registering task
> and sending final execution state FAILED to Job
> Manager for task Source: Custom Source 85cfdb83f536b26e07ca2aa4a1b66302.
>
>
>
>
>

[九思]

在TaskManager看到了这两个错误：
2019-07-22 05:39:03,987 WARN  org.apache.hadoop.ipc.Client                                  - Failed to connect to server: master/10.0.2.11:9000: try once and fail. java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:659) at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWithTimeout.java:192) at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:531) at org.apache.hadoop.ipc.Client$Connection.setupConnection(Client.java:685) at org.apache.hadoop.ipc.Client$Connection.setupIOstreams(Client.java:788) at org.apache.hadoop.ipc.Client$Connection.access$3500(Client.java:410) at org.apache.hadoop.ipc.Client.getConnection(Client.java:1550) at org.apache.hadoop.ipc.Client.call(Client.java:1381) at org.apache.hadoop.ipc.Client.call(Client.java:1345) at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:227) at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:116) at com.sun.proxy.$Proxy17.mkdirs(Unknown Source) at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.mkdirs(ClientNamenodeProtocolTranslatorPB.java:583) 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) at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:409) at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeMethod(RetryInvocationHandler.java:163) at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invoke(RetryInvocationHandler.java:155) at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeOnce(RetryInvocationHandler.java:95) at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:346) at com.sun.proxy.$Proxy18.mkdirs(Unknown Source) at org.apache.hadoop.hdfs.DFSClient.primitiveMkdir(DFSClient.java:2540) at org.apache.hadoop.hdfs.DFSClient.mkdirs(DFSClient.java:2515) at org.apache.hadoop.hdfs.DistributedFileSystem$25.doCall(DistributedFileSystem.java:1156) at org.apache.hadoop.hdfs.DistributedFileSystem$25.doCall(DistributedFileSystem.java:1153) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirsInternal(DistributedFileSystem.java:1153) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirs(DistributedFileSystem.java:1145) at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1914) at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.mkdirs(HadoopFileSystem.java:170) at org.apache.flink.core.fs.SafetyNetWrapperFileSystem.mkdirs(SafetyNetWrapperFileSystem.java:112) at org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage.<init>(MemoryBackendCheckpointStorage.java:87) at org.apache.flink.runtime.state.memory.MemoryStateBackend.createCheckpointStorage(MemoryStateBackend.java:295) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:257) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) at java.lang.Thread.run(Thread.java:745) .jar:/home/hadoop-2.9.0/share/hadoop/common/lib/jackson-jaxrs-1.9.13.jar:/home/hadoop-2.9.0/share/hadoop/common/lib/apacheds-kerberos-codec-2.0.0-M15.jar:/home/hadoop-2.9.0/share/hadoop/common/lib/guava-11.0.2.jar:/home/2019-07-22 05:39:02.960 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Creating FileSystem stream leak safety net for task Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2) (54be512ff5fc97d934ca9fbb96d66fe6) [DEPLOYING] 2019-07-22 05:39:02.960 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Loading JAR files for task Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2) (54be512ff5fc97d934ca9fbb96d66fe6) [DEPLOYING]. 2019-07-22 05:39:02.963 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Registering task at network: Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2) (54be512ff5fc97d934ca9fbb96d66fe6) [DEPLOYING]. 2019-07-22 05:39:02.963 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Registering task at network: Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2) (740a6bf3006287f24c46028abcbd8ec3) [DEPLOYING]. 2019-07-22 05:39:02.965 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Registering task at network: Source: Custom Source (1/1) (a556fff0d64b101420a4209c8ba7be7e) [DEPLOYING]. 2019-07-22 05:39:02.987 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2) (54be512ff5fc97d934ca9fbb96d66fe6) switched from DEPLOYING to RUNNING. 2019-07-22 05:39:02.997 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2)] INFO  org.apache.flink.runtime.taskmanager.Task  - Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2) (740a6bf3006287f24c46028abcbd8ec3) switched from DEPLOYING to RUNNING. 2019-07-22 05:39:03.013 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2)] INFO  org.apache.flink.streaming.runtime.tasks.StreamTask  - No state backend has been configured, using default (Memory / JobManager) MemoryStateBackend (data in heap memory / checkpoints to JobManager) (checkpoints: 'hdfs://10.0.2.11:9000/flink/igg-flink-log-monitor/checkpoints', savepoints: 'hdfs://10.0.2.11:9000/flink/igg-flink-log-monitor/savepoints', asynchronous: TRUE, maxStateSize: 5242880) 2019-07-22 05:39:03.020 [Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (2/2)] INFO  org.apache.flink.streaming.runtime.tasks.StreamTask  - No state backend has been configured, using default (Memory / JobManager) MemoryStateBackend (data in heap memory / checkpoints to JobManager) (checkpoints: 'hdfs://10.0.2.11:9000/flink/igg-flink-log-monitor/checkpoints', savepoints: 'hdfs://10.0.2.11:9000/flink/igg-flink-log-monitor/savepoints', asynchronous: TRUE, maxStateSize: 5242880) 2019-07-22 05:39:03.067 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Source: Custom Source (1/1) (a556fff0d64b101420a4209c8ba7be7e) switched from DEPLOYING to FAILED. java.lang.NullPointerException: null at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1175) at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212) at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:51) at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689) at java.lang.Thread.run(Thread.java:745) 2019-07-22 05:39:03.068 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Freeing task resources for Source: Custom Source (1/1) (a556fff0d64b101420a4209c8ba7be7e). 2019-07-22 05:39:03.090 [Source: Custom Source (1/1)] INFO  org.apache.flink.runtime.taskmanager.Task  - Ensuring all FileSystem streams are closed for task Source: Custom Source (1/1) (a556fff0d64b101420a4209c8ba7be7e) [FAILED] 2019-07-22 05:39:03.098 [flink-akka.actor.default-dispatcher-5] INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor  - Un-registering task and sending final execution state FAILED to JobManager for task Source: Custom Source a556fff0d64b101420a4209c8ba7be7e. 2019-07-22 05:39:03.176 [flink-akka.actor.default-dispatcher-5] INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor  - Discarding the results produced by task execution a556fff0d64b101420a4209c8ba7be7e. 2019-07-22 05:39:03.178 [flink-akka.actor.default-dispatcher-5] INFO  org.apache.flink.runtime.taskmanager.Task  - Attempting to cancel task Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2) (740a6bf3006287f24c46028abcbd8ec3). 2019-07-22 05:39:03.179 [flink-akka.actor.default-dispatcher-5] INFO  org.apache.flink.runtime.taskmanager.Task  - Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2) (740a6bf3006287f24c46028abcbd8ec3) switched from RUNNING to CANCELING. 2019-07-22 05:39:03.179 [flink-akka.actor.default-dispatcher-5] INFO  org.apache.flink.runtime.taskmanager.Task  - Triggering cancellation of task code Map -> Filter -> Timestamps/Watermarks -> Filter -> Sink: Unnamed (1/2) (740a6bf3006287f24c46028abcbd8ec3).




------------------ 原始邮件 ------------------
发件人: "zhisheng"<[hidden email]>;
发送时间: 2019年7月22日(星期一) 中午1:55
收件人: "user-zh"<[hidden email]>;

主题: Re: flink报空指针



hi 九思：
    从提供的日志来看是看不出什么问题导致的空指针异常，有更详细的日志或者把你的代码提供出来吗？或者可以简单参考我自己之前写的博客（写入数据到
rabbitmq）：http://1t.click/uh6
希望对你有帮助！

九思 <[hidden email]> 于2019年7月22日周一 下午1:32写道：

>  请教老师，flink on yarn，往rabbitmq写数据 或者
> 消费rabbitmq数据，都会报空指针，啥意思呢？本地idea都是正常的。flink包版本是1.8的，flink环境之前搭的1.7
>
> 2019-07-22 11:32:12.309 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Source: Custom Source (1/1)
> (85cfdb83f536b26e07ca2aa4a1b66302) switched from DEPLOYING
>  to FAILED.
> java.lang.NullPointerException: null
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1175)
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212)
>         at
> org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190)
>         at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:51)
>         at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native
> Method)
>         at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
>         at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>         at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>         at
> org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405)
>         at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689)
>         at java.lang.Thread.run(Thread.java:745)
> 2019-07-22 11:32:12.310 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Freeing task resources for
> Source: Custom Source (1/1) (85cfdb83f536b26e07ca2aa4a1b663
> 02).
> 2019-07-22 11:32:12.321 [Source: Custom Source (1/1)] INFO
> org.apache.flink.runtime.taskmanager.Task  - Ensuring all FileSystem
> streams are closed for task Source: Custom Source (1/1) (85cfd
> b83f536b26e07ca2aa4a1b66302) [FAILED]
> 2019-07-22 11:32:12.331 [flink-akka.actor.default-dispatcher-3] INFO
> org.apache.flink.runtime.taskexecutor.TaskExecutor  - Un-registering task
> and sending final execution state FAILED to Job
> Manager for task Source: Custom Source 85cfdb83f536b26e07ca2aa4a1b66302.
>
>
>
>
>