You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Piotr Nowojski (Jira)" <ji...@apache.org> on 2022/04/12 08:56:00 UTC
[jira] [Created] (FLINK-27202) NullPointerException on stop-with-savepoint with AsyncWaitOperator followed by FlinkKafkaProducer
Piotr Nowojski created FLINK-27202:
--------------------------------------
Summary: NullPointerException on stop-with-savepoint with AsyncWaitOperator followed by FlinkKafkaProducer
Key: FLINK-27202
URL: https://issues.apache.org/jira/browse/FLINK-27202
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka, Runtime / Task
Affects Versions: 1.13.6, 1.12.7
Reporter: Piotr Nowojski
Some lingering mails from {{AsyncWaitOperator}} (or other operators using mailbox, or maybe even processing time timers), that are chained with {{FlinkKafkaProducer}} can cause the following exceptions when using stop-with-savepoint:
{noformat}
2022-04-11 15:46:19,781 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - static enrichment -> Map -> Sink: enriched events sink (179/256) (3fefa588ad05fa8d2a10a6ad4a740cc6) switched from RUNNING to FAILED on 10.239.104.67:38149-12df6c @ 10.239.104.67 (dataPort=35745).
java.lang.NullPointerException: null
at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction$TransactionHolder.access$000(TwoPhaseCommitSinkFunction.java:591) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:223) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:50) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.emitResult(StreamRecordQueueEntry.java:64) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.queue.UnorderedStreamElementQueue$Segment.emitCompleted(UnorderedStreamElementQueue.java:272) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.queue.UnorderedStreamElementQueue.emitCompletedElement(UnorderedStreamElementQueue.java:159) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.outputCompletedElement(AsyncWaitOperator.java:287) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.access$100(AsyncWaitOperator.java:78) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.processResults(AsyncWaitOperator.java:356) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.lambda$processInMailbox$0(AsyncWaitOperator.java:337) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.drain(MailboxProcessor.java:170) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:647) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[flink-dist_2.12-1.12.7-stream2.jar:1.12.7-stream2]
at java.lang.Thread.run(Thread.java:829) ~[?:?]
{noformat}
This happens since {{FlinkKafkaProducer}} can be closed, without quiescing the mailbox. This issue might have been fixed by either FLINK-23532
or FLINK-23408.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)