xuyangzhong commented on code in PR #25892:
URL: https://github.com/apache/flink/pull/25892#discussion_r1922105929
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TableSinkITCase.scala:
##########
@@ -630,4 +635,66 @@ class TableSinkITCase(mode: StateBackendMode) extends
StreamingWithStateTestBase
"+I[3, clark, 1, null, null]")
assertThat(result.sorted).isEqualTo(expected.sorted)
}
+
+ @TestTemplate
+ def testUpsertSinkWithFailingSource(): Unit = {
+ // enable checkpoint, we are using failing source to force have a complete
checkpoint
+ // and cover restore path
+ env.enableCheckpointing(100, CheckpointingMode.EXACTLY_ONCE)
+ val configuration = new Configuration()
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "fixeddelay")
+
configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
Int.box(1))
+ configuration.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
+ Duration.ofMillis(0))
+ env.configure(configuration, Thread.currentThread.getContextClassLoader)
+ FailingCollectionSource.reset()
+
+ val data = List(
+ changelogRow("+I", Int.box(1), "Jim"),
+ changelogRow("-U", Int.box(1), "Jim"),
+ changelogRow("+U", Int.box(1), "Ketty"),
+ changelogRow("+I", Int.box(2), "Lilith"),
+ changelogRow("-U", Int.box(2), "Lilith"),
+ // failover
+ changelogRow("+I", Int.box(3), "Sam"),
+ changelogRow("-U", Int.box(3), "Sam"),
+ changelogRow("+U", Int.box(3), "Boob"),
+ changelogRow("-D", Int.box(3), "Boob"),
+ changelogRow("+I", Int.box(4), "Julia")
+ )
+ tEnv.executeSql(s"""
+ |CREATE TABLE pk_src (
+ | id int primary key not enforced,
+ | name string
+ |) with (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I,UA,UB,D',
+ | 'failing-source' = 'true',
+ | 'data-id' =
'${TestValuesTableFactory.registerData(data)}'
+ |)
+ |""".stripMargin)
+
+ tEnv.executeSql(s"""
+ |CREATE TABLE pk_snk (
+ | id int primary key not enforced,
+ | name string
+ |) with (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false',
+ | 'sink-changelog-mode-enforced' = 'I,UA,D'
+ |)
+ |""".stripMargin)
+
+ tEnv
+ .executeSql("""
+ |INSERT INTO pk_snk SELECT * FROM pk_src where name <>
'unknown';
+ |""".stripMargin)
+ .await()
+
+ val expected = List("+I[1, Ketty]", "+I[4, Julia]")
Review Comment:
@xishuaidelin Before this fix, the result is :
```
Actual :ArrayBuffer(+I[4, Julia], +U[1, Ketty], -U[2, Lilith])
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
