xuyangzhong commented on code in PR #25892:
URL: https://github.com/apache/flink/pull/25892#discussion_r1946286251
##########
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()
Review Comment:
I still recommend retaining FailingCollectionSource.reset(). Just like in
other tests that utilize `FailingCollectionSource`, where it is reset in the
`@BeforeEach` method to prevent interference from other tests, it should be
treated the same way here.
--
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
