xishuaidelin commented on code in PR #25892:
URL: https://github.com/apache/flink/pull/25892#discussion_r1922158356
##########
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:
Nit: I commented out this part and found that it has no impact. Could you
explain the specific reason for this configuration? Could we remove this?
--
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
