Here is the line where NPE was thrown: mainThread.interrupt(); // the main thread may be sleeping for the discovery interval
I wonder if runFetcher() encountered running being false - otherwise mainThread should not have been null. Looks like we should check whether mainThread is null when shutting down. On Thu, Apr 13, 2017 at 5:46 PM, Sathi Chowdhury < sathi.chowdh...@elliemae.com> wrote: > The taskmanger log does not point a line in my code ..but it seems like > the error occurs while it is trying to fetch kinesis record inside > connector jar > > > > red sequence number 495722619081512695413431879198 > 20576263466496304458235906 > > 2017-04-13 23:28:23,470 INFO org.apache.flink.streaming. > connectors.kinesis.FlinkKinesisConsumer - Subtask 0 is seeding the > fetcher with restored shard KinesisStreamShard{streamName= > 'dev-ingest-kinesis-us-west-2', shard='{ShardId: > shardId-000000000009,HashKeyRange: > {StartingHashKey: 306254130228844617117037146688591390310,EndingHashKey: > 340282366920938463463374607431768211455},SequenceNumberRange: > {StartingSequenceNumber: 495722540788279459864077892456 > 74345090539511066904232082,}}'}, starting state set to the restored > sequence number LATEST_SEQUENCE_NUM > > 2017-04-13 23:28:23,471 WARN org.apache.flink.streaming. > connectors.kinesis.FlinkKinesisConsumer - Error while closing Kinesis > data fetcher > > java.lang.NullPointerException > > at org.apache.flink.streaming.connectors.kinesis.internals. > KinesisDataFetcher.shutdownFetcher(KinesisDataFetcher.java:472) > > at org.apache.flink.streaming.connectors.kinesis. > FlinkKinesisConsumer.cancel(FlinkKinesisConsumer.java:246) > > at org.apache.flink.streaming.connectors.kinesis. > FlinkKinesisConsumer.close(FlinkKinesisConsumer.java:256) > > at org.apache.flink.api.common.functions.util.FunctionUtils. > closeFunction(FunctionUtils.java:43) > > at org.apache.flink.streaming.api.operators. > AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:127) > > at org.apache.flink.streaming.runtime.tasks.StreamTask. > disposeAllOperators(StreamTask.java:442) > > at org.apache.flink.streaming.runtime.tasks.StreamTask. > invoke(StreamTask.java:343) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:655) > > at java.lang.Thread.run(Thread.java:745) > > 2017-04-13 23:28:23,471 WARN org.apache.flink.streaming. > connectors.kinesis.FlinkKinesisConsumer - Error while closing Kinesis > data fetcher > > java.lang.NullPointerException > > at org.apache.flink.streaming.connectors.kinesis.internals. > KinesisDataFetcher.shutdownFetcher(KinesisDataFetcher.java:472) > > at org.apache.flink.streaming.connectors.kinesis. > FlinkKinesisConsumer.cancel(FlinkKinesisConsumer.java:246) > > at org.apache.flink.streaming.connectors.kinesis. > FlinkKinesisConsumer.close(FlinkKinesisConsumer.java:256) > > at org.apache.flink.api.common.functions.util.FunctionUtils. > closeFunction(FunctionUtils.java:43) > > at org.apache.flink.streaming.api.operators. > AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:127) > > at org.apache.flink.streaming.runtime.tasks.StreamTask. > disposeAllOperators(StreamTask.java:442) > > at org.apache.flink.streaming.runtime.tasks.StreamTask. > invoke(StreamTask.java:343) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:655) > > at java.lang.Thread.run(Thread.java:745) > > 2017-04-13 23:28:23,472 INFO org.apache.flink.runtime.taskmanager.Task > - Source: Custom Source (1/1) ( > 8a7301a437cb2d052208ee42c994104b) switched from CANCELING to CANCELED. > > 2017-04-13 23:28:23,471 INFO org.apache.flink.streaming. > connectors.kinesis.FlinkKinesisConsumer - Subtask 0 is seeding the > fetcher with restored shard KinesisStreamShard{streamName= > 'dev-ingest-kinesis-us-west-2', shard='{ShardId: > shardId-000000000006,HashKeyRange: > {StartingHashKey: 204169420152563078078024764459060926873,EndingHashKey: > 238197656844656924424362225202237748018},SequenceNumberRange: > {StartingSequenceNumber: 495722540787610437508121973762 > 49737935721565982386290786,}}'}, starting state set to the restored > sequence number LATEST_SEQUENCE_NUM > > 2017-04-13 23:28:23,472 INFO org.apache.flink.runtime. > taskmanager.Task - Freeing task resources for Source: > Custom Source (1/1) (8a7301a437cb2d052208ee42c994104b). > > > > > > *From: *Sathi Chowdhury <sathi.chowdh...@elliemae.com> > *Date: *Thursday, April 13, 2017 at 5:44 PM > *To: *Ted Yu <yuzhih...@gmail.com> > > *Cc: *"user@flink.apache.org" <user@flink.apache.org> > *Subject: *Re: Flink errors out and job fails--IOException from > CollectSink.open() > > > > Hi Ted, Sorry for my big font earlier…was not intended J > > > > I am on flink 1.2.0 > > I built flink-connector-kinesis_2.10-1.2.0.jar from source and included > in the fatjar I am running. > > Followed this http://www.kidder.io/2017/02/15/flink-kinesis-streaming- > connector/ > <https://na01.safelinks.protection.outlook.com/?url=http%3A%2F%2Fwww.kidder.io%2F2017%2F02%2F15%2Fflink-kinesis-streaming-connector%2F&data=01%7C01%7C%7Cfc97be832d1a4062d75608d482cf675f%7C0d009d13c2cd47d891dd2ae838b00d4b%7C0&sdata=avPl4%2FU5DTjZW09Zby9CeUttUNpiGWH%2Bvnycy9PhUDA%3D&reserved=0> > > > > > > From code I read a kinesis stream using > > > > consumerConfig.setProperty(ConsumerConfigConstants.*AWS_REGION*, region); > consumerConfig.setProperty(ConsumerConfigConstants. > *DEFAULT_STREAM_INITIAL_POSITION*, "LATEST"); > consumerConfig.setProperty(ConsumerConfigConstants. > *AWS_CREDENTIALS_PROVIDER*, "AUTO"); > consumerConfig.setProperty(ConsumerConfigConstants.*SHARD_GETRECORDS_MAX*, > "10"); > consumerConfig.setProperty(ConsumerConfigConstants. > *SHARD_GETRECORDS_RETRIES*, "200"); > consumerConfig.setProperty(ConsumerConfigConstants. > *SHARD_GETRECORDS_INTERVAL_MILLIS*, "2000"); > DataStream<Map<String, Object>> stream = env.addSource(new > FlinkKinesisConsumer<>(inputStream, new MyJsonDeserializationSchema(), > consumerConfig)); > > > > > > While I push the json event to the Kinesis stream intermittently I see > this NPE and flink job fails > > > > 2017-04-14 00:31:54,672 WARN > org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer - Error > while closing Kinesis data fetcher > > java.lang.NullPointerException > > at > org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.shutdownFetcher(KinesisDataFetcher.java:472) > > at > org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.cancel(FlinkKinesisConsumer.java:246) > > at > org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.close(FlinkKinesisConsumer.java:256) > > at > org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43) > > at > org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:127) > > at > org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:442) > > at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:343) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:655) > > at java.lang.Thread.run(Thread.java:745) > > > > Thanks > > Sathi > > > > > > > > *From: *Ted Yu <yuzhih...@gmail.com> > *Date: *Thursday, April 13, 2017 at 5:02 PM > *To: *Sathi Chowdhury <sathi.chowdh...@elliemae.com> > *Cc: *"user@flink.apache.org" <user@flink.apache.org> > *Subject: *Re: Flink errors out and job fails--IOException from > CollectSink.open() > > > > Can you give us a bit more information ? > > > > release of flink > > snippet of your code > > > > Thanks > > =============Notice to Recipient: This e-mail transmission, and any > documents, files or previous e-mail messages attached to it may contain > information that is confidential or legally privileged, and intended for > the use of the individual or entity named above. If you are not the > intended recipient, or a person responsible for delivering it to the > intended recipient, you are hereby notified that you must not read this > transmission and that any disclosure, copying, printing, distribution or > use of any of the information contained in or attached to this transmission > is STRICTLY PROHIBITED. If you have received this transmission in error, > please immediately notify the sender by telephone or return e-mail and > delete the original transmission and its attachments without reading or > saving in any manner. Thank you. ============= > =============Notice to Recipient: This e-mail transmission, and any > documents, files or previous e-mail messages attached to it may contain > information that is confidential or legally privileged, and intended for > the use of the individual or entity named above. If you are not the > intended recipient, or a person responsible for delivering it to the > intended recipient, you are hereby notified that you must not read this > transmission and that any disclosure, copying, printing, distribution or > use of any of the information contained in or attached to this transmission > is STRICTLY PROHIBITED. If you have received this transmission in error, > please immediately notify the sender by telephone or return e-mail and > delete the original transmission and its attachments without reading or > saving in any manner. Thank you. ============= >