Skip to content

Commit 6daa30f

Browse files
lincoln-lillincoln-lil
authored
[FLINK-22604][table-runtime-blink] Fix NPE on bundle close when task failover after a failed task open
This closes #15863
1 parent fbf84ac commit 6daa30f

7 files changed

Lines changed: 256 additions & 103 deletions

File tree

‎flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/GroupAggregateHarnessTest.scala‎

Lines changed: 46 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -19,16 +19,20 @@
1919
package org.apache.flink.table.planner.runtime.harness
2020

2121
import org.apache.flink.api.scala._
22-
import org.apache.flink.table.api.{EnvironmentSettings, _}
22+
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness
2323
import org.apache.flink.table.api.bridge.scala._
2424
import org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl
2525
import org.apache.flink.table.api.config.ExecutionConfigOptions.{TABLE_EXEC_MINIBATCH_ALLOW_LATENCY, TABLE_EXEC_MINIBATCH_ENABLED, TABLE_EXEC_MINIBATCH_SIZE}
2626
import org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY
27+
import org.apache.flink.table.api.{EnvironmentSettings, _}
28+
import org.apache.flink.table.data.RowData
2729
import org.apache.flink.table.planner.runtime.utils.StreamingWithMiniBatchTestBase.{MiniBatchMode, MiniBatchOff, MiniBatchOn}
2830
import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.{HEAP_BACKEND, ROCKSDB_BACKEND, StateBackendMode}
2931
import org.apache.flink.table.planner.runtime.utils.UserDefinedFunctionTestUtils.CountNullNonNull
32+
import org.apache.flink.table.runtime.typeutils.RowDataSerializer
3033
import org.apache.flink.table.runtime.util.RowDataHarnessAssertor
3134
import org.apache.flink.table.runtime.util.StreamRecordUtils.binaryRecord
35+
import org.apache.flink.table.types.logical.LogicalType
3236
import org.apache.flink.types.Row
3337
import org.apache.flink.types.RowKind._
3438

@@ -46,7 +50,7 @@ import scala.collection.mutable
4650

4751
@RunWith(classOf[Parameterized])
4852
class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode)
49-
extends HarnessTestBase(mode) {
53+
extends HarnessTestBase(mode) {
5054

5155
@Before
5256
override def before(): Unit = {
@@ -100,7 +104,7 @@ class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode
100104
testHarness.setStateTtlProcessingTime(1)
101105

102106
// insertion
103-
testHarness.processElement(binaryRecord(INSERT,"aaa", 1L: JLong))
107+
testHarness.processElement(binaryRecord(INSERT, "aaa", 1L: JLong))
104108
expectedOutput.add(binaryRecord(INSERT, "aaa", 1L: JLong))
105109

106110
// insertion
@@ -144,7 +148,7 @@ class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode
144148
expectedOutput.add(binaryRecord(INSERT, "eee", 6L: JLong))
145149

146150
// retract
147-
testHarness.processElement(binaryRecord(INSERT,"aaa", 7L: JLong))
151+
testHarness.processElement(binaryRecord(INSERT, "aaa", 7L: JLong))
148152
expectedOutput.add(binaryRecord(UPDATE_BEFORE, "aaa", 9L: JLong))
149153
expectedOutput.add(binaryRecord(UPDATE_AFTER, "aaa", 16L: JLong))
150154
testHarness.processElement(binaryRecord(INSERT, "bbb", 3L: JLong))
@@ -160,28 +164,8 @@ class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode
160164

161165
@Test
162166
def testAggregationWithDistinct(): Unit = {
163-
val data = new mutable.MutableList[(String, String, Long)]
164-
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
165-
tEnv.createTemporaryView("T", t)
166-
tEnv.createTemporarySystemFunction("CntNullNonNull", new CountNullNonNull)
167-
168-
val sql =
169-
"""
170-
|SELECT a, COUNT(DISTINCT b), CntNullNonNull(DISTINCT b), COUNT(*), SUM(c)
171-
|FROM T
172-
|GROUP BY a
173-
""".stripMargin
174-
val t1 = tEnv.sqlQuery(sql)
175-
176-
tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(2))
177-
val testHarness = createHarnessTester(t1.toRetractStream[Row], "GroupAggregate")
178-
val assertor = new RowDataHarnessAssertor(
179-
Array(
180-
DataTypes.STRING().getLogicalType,
181-
DataTypes.BIGINT().getLogicalType,
182-
DataTypes.STRING().getLogicalType,
183-
DataTypes.BIGINT().getLogicalType,
184-
DataTypes.BIGINT().getLogicalType))
167+
val (testHarness, outputTypes) = createAggregationWithDistinct
168+
val assertor = new RowDataHarnessAssertor(outputTypes)
185169

186170
testHarness.open()
187171

@@ -191,7 +175,7 @@ class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode
191175
testHarness.setStateTtlProcessingTime(1)
192176

193177
// insertion
194-
testHarness.processElement(binaryRecord(INSERT,"aaa", "a1", 1L: JLong))
178+
testHarness.processElement(binaryRecord(INSERT, "aaa", "a1", 1L: JLong))
195179
expectedOutput.add(binaryRecord(INSERT, "aaa", 1L: JLong, "1|0", 1L: JLong, 1L: JLong))
196180

197181
// insertion
@@ -240,6 +224,41 @@ class GroupAggregateHarnessTest(mode: StateBackendMode, miniBatch: MiniBatchMode
240224
testHarness.close()
241225
}
242226

227+
private def createAggregationWithDistinct()
228+
: (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], Array[LogicalType]) = {
229+
val data = new mutable.MutableList[(String, String, Long)]
230+
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
231+
tEnv.createTemporaryView("T", t)
232+
tEnv.createTemporarySystemFunction("CntNullNonNull", new CountNullNonNull)
233+
234+
val sql =
235+
"""
236+
|SELECT a, COUNT(DISTINCT b), CntNullNonNull(DISTINCT b), COUNT(*), SUM(c)
237+
|FROM T
238+
|GROUP BY a
239+
""".stripMargin
240+
val t1 = tEnv.sqlQuery(sql)
241+
242+
tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(2))
243+
val testHarness = createHarnessTester(t1.toRetractStream[Row], "GroupAggregate")
244+
val outputTypes = Array(
245+
DataTypes.STRING().getLogicalType,
246+
DataTypes.BIGINT().getLogicalType,
247+
DataTypes.STRING().getLogicalType,
248+
DataTypes.BIGINT().getLogicalType,
249+
DataTypes.BIGINT().getLogicalType)
250+
251+
(testHarness, outputTypes)
252+
}
253+
254+
@Test
255+
def testCloseWithoutOpen(): Unit = {
256+
val (testHarness, outputType) = createAggregationWithDistinct
257+
testHarness.setup(new RowDataSerializer(outputType: _*))
258+
// simulate a failover after a failed task open(e.g., stuck on initializing)
259+
// expect no exception happens
260+
testHarness.close()
261+
}
243262
}
244263

245264
object GroupAggregateHarnessTest {

‎flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/HarnessTestBase.scala‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,9 @@ import org.apache.flink.runtime.state.StateBackend
2828
import org.apache.flink.runtime.state.memory.MemoryStateBackend
2929
import org.apache.flink.streaming.api.operators.OneInputStreamOperator
3030
import org.apache.flink.streaming.api.scala.DataStream
31-
import org.apache.flink.streaming.api.transformations.OneInputTransformation
31+
import org.apache.flink.streaming.api.transformations.{OneInputTransformation, PartitionTransformation}
3232
import org.apache.flink.streaming.api.watermark.Watermark
33-
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness
33+
import org.apache.flink.streaming.util.{KeyedOneInputStreamOperatorTestHarness, OneInputStreamOperatorTestHarness}
3434
import org.apache.flink.table.data.RowData
3535
import org.apache.flink.table.planner.JLong
3636
import org.apache.flink.table.planner.runtime.utils.StreamingTestBase
@@ -86,6 +86,19 @@ class HarnessTestBase(mode: StateBackendMode) extends StreamingTestBase {
8686
.asInstanceOf[KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData]]
8787
}
8888

89+
def createHarnessTesterForNoState(
90+
ds: DataStream[_],
91+
prefixOperatorName: String)
92+
: OneInputStreamOperatorTestHarness[RowData, RowData] = {
93+
val transformation = extractExpectedTransformation(
94+
ds.javaStream.getTransformation,
95+
prefixOperatorName)
96+
val processOperator = transformation.getOperator
97+
.asInstanceOf[OneInputStreamOperator[Any, Any]]
98+
new OneInputStreamOperatorTestHarness(processOperator)
99+
.asInstanceOf[OneInputStreamOperatorTestHarness[RowData, RowData]]
100+
}
101+
89102
private def extractExpectedTransformation(
90103
t: Transformation[_],
91104
prefixOperatorName: String): OneInputTransformation[_, _] = {
@@ -96,6 +109,8 @@ class HarnessTestBase(mode: StateBackendMode) extends StreamingTestBase {
96109
} else {
97110
extractExpectedTransformation(one.getInputs.get(0), prefixOperatorName)
98111
}
112+
case p: PartitionTransformation[_] =>
113+
extractExpectedTransformation(p.getInputs.get(0), prefixOperatorName)
99114
case _ => throw new Exception(
100115
s"Can not find the expected $prefixOperatorName transformation")
101116
}

‎flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverAggregateHarnessTest.scala‎

Lines changed: 44 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,17 @@ package org.apache.flink.table.planner.runtime.harness
2020

2121
import org.apache.flink.api.common.time.Time
2222
import org.apache.flink.api.scala._
23-
import org.apache.flink.streaming.api.TimeCharacteristic
2423
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
24+
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness
2525
import org.apache.flink.table.api._
2626
import org.apache.flink.table.api.bridge.scala._
2727
import org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl
28+
import org.apache.flink.table.data.RowData
2829
import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.StateBackendMode
30+
import org.apache.flink.table.runtime.typeutils.RowDataSerializer
2931
import org.apache.flink.table.runtime.util.RowDataHarnessAssertor
3032
import org.apache.flink.table.runtime.util.StreamRecordUtils.{binaryrow, row}
33+
import org.apache.flink.table.types.logical.LogicalType
3134
import org.apache.flink.types.Row
3235

3336
import org.junit.runner.RunWith
@@ -52,34 +55,8 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
5255

5356
@Test
5457
def testProcTimeBoundedRowsOver(): Unit = {
55-
56-
val data = new mutable.MutableList[(Long, String, Long)]
57-
val t = env.fromCollection(data).toTable(tEnv, 'currtime, 'b, 'c, 'proctime.proctime)
58-
tEnv.registerTable("T", t)
59-
60-
val sql =
61-
"""
62-
|SELECT currtime, b, c,
63-
| min(c) OVER
64-
| (PARTITION BY b ORDER BY proctime ROWS BETWEEN 1 PRECEDING AND CURRENT ROW),
65-
| max(c) OVER
66-
| (PARTITION BY b ORDER BY proctime ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
67-
|FROM T
68-
""".stripMargin
69-
val t1 = tEnv.sqlQuery(sql)
70-
71-
tEnv.getConfig.setIdleStateRetentionTime(Time.seconds(2), Time.seconds(4))
72-
val testHarness = createHarnessTester(t1.toAppendStream[Row], "OverAggregate")
73-
val assertor = new RowDataHarnessAssertor(
74-
Array(
75-
DataTypes.BIGINT().getLogicalType,
76-
DataTypes.STRING().getLogicalType,
77-
DataTypes.BIGINT().getLogicalType,
78-
DataTypes.BIGINT().getLogicalType,
79-
DataTypes.BIGINT().getLogicalType,
80-
DataTypes.BIGINT().getLogicalType,
81-
DataTypes.BIGINT().getLogicalType))
82-
58+
val (testHarness, outputType) = createProcTimeBoundedRowsOver
59+
val assertor = new RowDataHarnessAssertor(outputType)
8360
testHarness.open()
8461

8562
// register cleanup timer with 3001
@@ -161,6 +138,36 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
161138
testHarness.close()
162139
}
163140

141+
private def createProcTimeBoundedRowsOver()
142+
: (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], Array[LogicalType]) = {
143+
val data = new mutable.MutableList[(Long, String, Long)]
144+
val t = env.fromCollection(data).toTable(tEnv, 'currtime, 'b, 'c, 'proctime.proctime)
145+
tEnv.registerTable("T", t)
146+
147+
val sql =
148+
"""
149+
|SELECT currtime, b, c,
150+
| min(c) OVER
151+
| (PARTITION BY b ORDER BY proctime ROWS BETWEEN 1 PRECEDING AND CURRENT ROW),
152+
| max(c) OVER
153+
| (PARTITION BY b ORDER BY proctime ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
154+
|FROM T
155+
""".stripMargin
156+
val t1 = tEnv.sqlQuery(sql)
157+
158+
tEnv.getConfig.setIdleStateRetentionTime(Time.seconds(2), Time.seconds(4))
159+
val testHarness = createHarnessTester(t1.toAppendStream[Row], "OverAggregate")
160+
val outputType = Array(
161+
DataTypes.BIGINT().getLogicalType,
162+
DataTypes.STRING().getLogicalType,
163+
DataTypes.BIGINT().getLogicalType,
164+
DataTypes.BIGINT().getLogicalType,
165+
DataTypes.BIGINT().getLogicalType,
166+
DataTypes.BIGINT().getLogicalType,
167+
DataTypes.BIGINT().getLogicalType)
168+
(testHarness, outputType)
169+
}
170+
164171
/**
165172
* NOTE: all elements at the same proc timestamp have the same value per key
166173
*/
@@ -940,4 +947,12 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
940947
assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
941948
testHarness.close()
942949
}
950+
951+
@Test
952+
def testCloseWithoutOpen(): Unit = {
953+
val (testHarness, outputType) = createProcTimeBoundedRowsOver
954+
testHarness.setup(new RowDataSerializer(outputType: _*))
955+
// simulate a failover after a failed task open, expect no exception happens
956+
testHarness.open()
957+
}
943958
}

‎flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/TableAggregateHarnessTest.scala‎

Lines changed: 38 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -18,23 +18,27 @@
1818

1919
package org.apache.flink.table.planner.runtime.harness
2020

21-
import java.lang.{Integer => JInt}
22-
import java.util.concurrent.ConcurrentLinkedQueue
2321
import org.apache.flink.api.scala._
24-
import org.apache.flink.table.api._
22+
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness
2523
import org.apache.flink.table.api.bridge.scala._
2624
import org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl
27-
import org.apache.flink.table.api.EnvironmentSettings
25+
import org.apache.flink.table.api.{EnvironmentSettings, _}
26+
import org.apache.flink.table.data.RowData
2827
import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.StateBackendMode
2928
import org.apache.flink.table.planner.utils.{Top3WithMapView, Top3WithRetractInput}
29+
import org.apache.flink.table.runtime.typeutils.RowDataSerializer
3030
import org.apache.flink.table.runtime.util.RowDataHarnessAssertor
3131
import org.apache.flink.table.runtime.util.StreamRecordUtils.{deleteRecord, insertRecord}
32+
import org.apache.flink.table.types.logical.LogicalType
3233
import org.apache.flink.types.Row
34+
3335
import org.junit.runner.RunWith
3436
import org.junit.runners.Parameterized
3537
import org.junit.{Before, Test}
3638

39+
import java.lang.{Integer => JInt}
3740
import java.time.Duration
41+
import java.util.concurrent.ConcurrentLinkedQueue
3842

3943
import scala.collection.mutable
4044

@@ -117,22 +121,8 @@ class TableAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(
117121

118122
@Test
119123
def testTableAggregateWithRetractInput(): Unit = {
120-
val top3 = new Top3WithRetractInput
121-
tEnv.registerFunction("top3", top3)
122-
val source = env.fromCollection(data).toTable(tEnv, 'a, 'b)
123-
val resultTable = source
124-
.groupBy('a)
125-
.select('b.sum as 'b)
126-
.flatAggregate(top3('b) as ('b1, 'b2))
127-
.select('b1, 'b2)
128-
129-
tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(2))
130-
val testHarness = createHarnessTester(
131-
resultTable.toRetractStream[Row], "GroupTableAggregate")
132-
val assertor = new RowDataHarnessAssertor(
133-
Array(
134-
DataTypes.INT().getLogicalType,
135-
DataTypes.INT().getLogicalType))
124+
val (testHarness, outputTypes) = createTableAggregateWithRetract
125+
val assertor = new RowDataHarnessAssertor(outputTypes)
136126

137127
testHarness.open()
138128
val expectedOutput = new ConcurrentLinkedQueue[Object]()
@@ -170,4 +160,32 @@ class TableAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(
170160
assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
171161
testHarness.close()
172162
}
163+
164+
private def createTableAggregateWithRetract()
165+
: (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], Array[LogicalType]) = {
166+
val top3 = new Top3WithRetractInput
167+
tEnv.registerFunction("top3", top3)
168+
val source = env.fromCollection(data).toTable(tEnv, 'a, 'b)
169+
val resultTable = source
170+
.groupBy('a)
171+
.select('b.sum as 'b)
172+
.flatAggregate(top3('b) as('b1, 'b2))
173+
.select('b1, 'b2)
174+
175+
tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(2))
176+
val testHarness = createHarnessTester(
177+
resultTable.toRetractStream[Row], "GroupTableAggregate")
178+
val outputTypes = Array(
179+
DataTypes.INT().getLogicalType,
180+
DataTypes.INT().getLogicalType)
181+
(testHarness, outputTypes)
182+
}
183+
184+
@Test
185+
def testCloseWithoutOpen(): Unit = {
186+
val (testHarness, outputTypes) = createTableAggregateWithRetract
187+
testHarness.setup(new RowDataSerializer(outputTypes: _*))
188+
// simulate a failover after a failed task open, expect no exception happens
189+
testHarness.close()
190+
}
173191
}

0 commit comments

Comments
 (0)