[jira] [Created] (FLINK-11568) Exception in Kinesis ShardConsumer hidden by InterruptedException

[Shang Yuanchun (Jira)]

Shannon Carey created FLINK-11568:
-------------------------------------

             Summary: Exception in Kinesis ShardConsumer hidden by InterruptedException
                 Key: FLINK-11568
                 URL: https://issues.apache.org/jira/browse/FLINK-11568
             Project: Flink
          Issue Type: Improvement
          Components: Kinesis Connector
    Affects Versions: 1.6.2
            Reporter: Shannon Carey
            Assignee: Shannon Carey


When the Kinesis ShardConsumer encounters an exception, for example due to a problem in the Deserializer, the root cause exception is often hidden by a non-informative InterruptedException caused by the FlinkKinesisConsumer thread being interrupted.

Ideally, the root cause exception would be preserved and thrown so that the logs contain enough information to diagnose the issue.

This probably affects all versions.

Here's an example of a log message with the unhelpful InterruptedException:
{code:java}
2019-02-05 13:29:31:383 thread=Source: Custom Source -> Filter -> Map -> Sink: Unnamed (1/8), level=WARN, logger=org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer, message="Error while closing Kinesis data fetcher"
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method)
at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.awaitTermination(KinesisDataFetcher.java:450)
at org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.cancel(FlinkKinesisConsumer.java:314)
at org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.close(FlinkKinesisConsumer.java:323)
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43)
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117)
at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:745)
{code}
And here's an example of the real exception that we're actually interested in, which is stored inside KinesisDataFetcher#error, but is not thrown or logged:
{code:java}
org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:416)
org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:290)
org.apache.avro.io.parsing.Parser.advance(Parser.java:88)
org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:267)
org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:178)
org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:240)
org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:230)
org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:174)
org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)
org.apache.flink.formats.avro.AvroDeserializationSchema.deserialize(AvroDeserializationSchema.java:135)
org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchemaWrapper.deserialize(KinesisDeserializationSchemaWrapper.java:44)
org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.deserializeRecordForCollectionAndUpdateState(ShardConsumer.java:332)
org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.run(ShardConsumer.java:231)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
java.util.concurrent.FutureTask.run(FutureTask.java)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
java.lang.Thread.run(Thread.java:745)
{code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)