[FLINK-22604] NPE on bundle close when task failover after a failed task open

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.13.0, 1.12.3
• Fix Version/s: 1.14.0
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: lincoln lee
• Reporter: lincoln lee
• Created: 08/May/21 05:45
• Updated: 28/Aug/21 11:19
• Resolved: 10/May/21 12:56

Description

We observed a NullPointerException when job failover after a failed task open(e.g., stuck on initializing), this should be avoid.

A simulated case will reproduce the exception:

@Test
def testCloseWithoutOpen(): Unit = {
    val (testHarness, outputType) = createAggregationWithDistinct  // see detail in the patch
    testHarness.setup(new RowDataSerializer(outputType: _*))
    // simulate a failover after a failed task open(e.g., stuck on initializing)
    // expect no exception happens
    testHarness.close()
}

exception:

java.lang.NullPointerExceptionjava.lang.NullPointerException at org.apache.flink.table.runtime.operators.bundle.AbstractMapBundleOperator.finishBundle(AbstractMapBundleOperator.java:134) at org.apache.flink.table.runtime.operators.bundle.AbstractMapBundleOperator.close(AbstractMapBundleOperator.java:156) at org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness.close(AbstractStreamOperatorTestHarness.java:691)

 

 

Issue Links

• links to: GitHub Pull Request #15863