Flink YARN Job unexpectedly switched to SUCCEEDED final status after CSV deserialization failure

[DONG, Weike]

Hi community,

I have noticed that when a wrong CSV record is ingested and deserialized at
CsvRowDeserializationSchema, when ignoreParseErrors is set to false, then
an IOException is thrown, which is expected, and in earlier Flink versions,
the Flink YARN app would terminated with FINISHED state and FAILED final
status for per-job mode Flink jobs without specifically setting a
RestartStrategy.

However, to my surprise, in Flink 1.10, the YARN app switched to SUCCEEDED
final status afterwards, which is weird since that this Flink job
termination was caused by abnormal data input, instead of normal end of
source stream.

Here I am confused because normally SUCCEEDED final status would give users
implications that the job has been **successfully** ended, however this is
definitely not the case. So I just wonder if this is the designed behavior
of Flink 1.10 on data deserialization errors.

Thank you for reading : )

Sincerely,
Weike

[DONG, Weike]

Hi dev,

After debugging and analysis, eventually the culprit is found. 

In order to cope with occasional data-format errors and intermittent source exceptions (Internet connection is somewhat unstable in my environment) that interrupt my Flink jobs occasionally, I have added a broad catch block in org.apache.flink.streaming.api.operators.StreamSource#run method to catch any exceptions thrown in the source operator by userFunction.run(ctx);, and this patch works well and keeps the jobs running smoothly without having to restart the job again and again. 

However, when the IOException is thrown at KafkaFetcher#run, in this method no catch block is defined, thus the exception would go directly to the finally block, which calls consumerThread.shutdown();. Therefore, even when this exception is later caught in StreamSource, it already causes KafkaFetcher to permanently close the consumer thread without any logging or warnings, so Flink can do nothing but finish the job.

In conclusion, it is my fault not to catch the exception in KafkaFetcher, and Flink is not the one to blame : )

Thanks

Sincerely,

Weike

On Tue, Mar 17, 2020 at 11:44 AM DONG, Weike <[hidden email]> wrote:

Hi community,

I have noticed that when a wrong CSV record is ingested and deserialized at CsvRowDeserializationSchema, when ignoreParseErrors is set to false, then an IOException is thrown, which is expected, and in earlier Flink versions, the Flink YARN app would terminated with FINISHED state and FAILED final status for per-job mode Flink jobs without specifically setting a RestartStrategy.

However, to my surprise, in Flink 1.10, the YARN app switched to SUCCEEDED final status afterwards, which is weird since that this Flink job termination was caused by abnormal data input, instead of normal end of source stream.

Here I am confused because normally SUCCEEDED final status would give users implications that the job has been **successfully** ended, however this is definitely not the case. So I just wonder if this is the designed behavior of Flink 1.10 on data deserialization errors.

Thank you for reading : )

Sincerely,

Weike