[FLINK-27760] NPE is thrown when executing PyFlink jobs in batch mode

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.13.0
• Fix Version/s: 1.14.5, 1.15.1, 1.16.0, 1.13.7
• Component/s: API / Python
• Labels: pull-request-available
• Assignee: Dian Fu
• Reporter: Dian Fu
• Created: 25/May/22 03:51
• Updated: 26/May/22 01:10
• Resolved: 26/May/22 01:10

Description

This is the exception stack reported by one user:

022-05-25 11:39:32,792 [MultipleInput(readOrder=[2,1,0], members=[\nHashJoin(joinType=[I... -> PythonCal ... with job vertex id 71d9b8e1b249eaa7e67ef93fb483177f (63/100)#123] WARN org.apache.flink.runtime.taskmanager.Task [] - MultipleInput(readOrder=[2,1,0], members=[\nHashJoin(joinType=[I... -> PythonCal ... with job vertex id 71d9b8e1b249eaa7e67ef93fb483177f (63/100)#123 (6fa78755ff19ac9d0d57aba21840e834) switched from INITIALIZING to FAILED with failure cause: java.lang.NullPointerException
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.getKeyedStateBackend(AbstractStreamOperator.java:466)
at org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.processElementsOfCurrentKeyIfNeeded(AbstractPythonFunctionOperator.java:238)
at org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.processWatermark(AbstractPythonFunctionOperator.java:208)
at org.apache.flink.table.runtime.operators.multipleinput.output.OneInputStreamOperatorOutput.emitWatermark(OneInputStreamOperatorOutput.java:45)
at org.apache.flink.streaming.api.operators.CountingOutput.emitWatermark(CountingOutput.java:39)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark(AbstractStreamOperator.java:632)
at org.apache.flink.table.runtime.operators.TableStreamOperator.processWatermark(TableStreamOperator.java:74)
at org.apache.flink.table.runtime.operators.multipleinput.output.OneInputStreamOperatorOutput.emitWatermark(OneInputStreamOperatorOutput.java:45)
at org.apache.flink.streaming.api.operators.CountingOutput.emitWatermark(CountingOutput.java:39)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark(AbstractStreamOperator.java:632)
at org.apache.flink.table.runtime.operators.TableStreamOperator.processWatermark(TableStreamOperator.java:74)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark2(AbstractStreamOperator.java:649)
at org.apache.flink.table.runtime.operators.multipleinput.input.SecondInputOfTwoInput.processWatermark(SecondInputOfTwoInput.java:44)
at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessorFactory$StreamTaskNetworkOutput.emitWatermark(StreamMultipleInputProcessorFactory.java:318)
at org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.findAndOutputNewMinWatermarkAcrossAlignedChannels(StatusWatermarkValve.java:196)
at org.apache.flink.streaming.runtime.streamstatus.StatusWatermarkValve.inputWatermark(StatusWatermarkValve.java:105)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:137)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:106)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessor.processInput(StreamMultipleInputProcessor.java:87)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:424)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:569)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:651)
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:541)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:586)
at java.lang.Thread.run(Thread.java:877)

Issue Links

• links to: GitHub Pull Request #19816