[jira] [Updated] (FLINK-30224) Add IT Case for NPE in FlinkKinesisConsumer's close() method

["Danny Cranmer (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-30224?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Danny Cranmer updated FLINK-30224:
----------------------------------
    Affects Version/s:     (was: 1.15.2)

> Add IT Case for NPE in FlinkKinesisConsumer's close() method
> ------------------------------------------------------------
>
>                 Key: FLINK-30224
>                 URL: https://issues.apache.org/jira/browse/FLINK-30224
>             Project: Flink
>          Issue Type: Improvement
>          Components: Connectors / Kinesis
>            Reporter: Astamur Kirillin
>            Assignee: Astamur Kirillin
>            Priority: Minor
>              Labels: pull-request-available
>
> In our project we observed an issue during too early call of {{stop-with-savepoint}} operation when we're calling it after 4-6 seconds after a job was started.
> We use Kinesis EFO consumer in our application and it looks like that it takes some time for the {{KinesisDataFetcher}} to be initialized and because of that {{FlinkKinesisConsumer#run(SourceContext<T>)}} method is slow. In some tests we call {{stop-with-savepoint}} after 4-6 seconds after a job was started and we have {{NullPointerException}} in {{{}FlinkKinesisConsumer#close(){}}}:
> {code:java}
> java.lang.NullPointerException: null
>         at org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.close(FlinkKinesisConsumer.java:421) ~[classes/:?]
>         at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) ~[classes/:?]
>         at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:114) ~[classes/:?]
>         at org.apache.flink.streaming.api.operators.StreamSource.close(StreamSource.java:124) ~[classes/:?]
>         at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163) ~[classes/:?]
>         at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125) ~[classes/:?]
>         at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:997) ~[classes/:?]
>         at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:876) ~[classes/:?]
>         at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:759) ~[classes/:?]
>         at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953) ~[classes/:?]
>         at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) ~[classes/:?]
>         at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746) ~[classes/:?]
>         at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568) ~[classes/:?]
>         at java.lang.Thread.run(Thread.java:829) ~[?:?] {code}
> We were able to reproduce this issue in {{FlinkKinesisITCase}} even for a standard KDS consumer. A fix is just an additional non-null check during consumer's close. We'll also add an IT-test for that.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)