fhueske commented on code in PR #28713:
URL: https://github.com/apache/flink/pull/28713#discussion_r3613640531
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java:
##########
@@ -491,6 +504,31 @@ void testRestore(TableTestProgram program, Path planPath,
String savepointPath)
}
}
+ /**
+ * Triggers a stop-with-savepoint, retrying while the job is not yet fully
running. A savepoint
+ * gated on source emission (see {@link #awaitSavepointReady}) can be
requested before all
+ * downstream tasks have reached RUNNING, which aborts with "Not all
required tasks are
+ * currently running". Sink-gated triggers never hit this (sink output
implies a running
+ * pipeline), so they succeed on the first attempt.
+ */
+ private String stopWithSavepointWhenRunning(JobClient jobClient) throws
Exception {
+ final long deadline = System.currentTimeMillis() +
RESULT_AWAIT_TIMEOUT_MILLIS;
+ while (true) {
+ try {
+ return jobClient
+ .stopWithSavepoint(false, tmpDir.toString(),
SavepointFormatType.DEFAULT)
+ .get();
+ } catch (ExecutionException e) {
+ final boolean notAllRunning =
+ e.getMessage() != null && e.getMessage().contains("Not
all required tasks");
+ if (!notAllRunning || System.currentTimeMillis() >= deadline) {
+ throw e;
+ }
Review Comment:
Before, the savepoint condition was reached when the expected "before
restore" rows were received by the sink. This implied that the whole pipeline
is running.
With the changes in this PR, we can take a savepoint after a certain number
of rows was emitted by the sources. This might also happen before all nodes are
running, hence this gate was added.
I've run the tests 20+ times and didn't observe any flakiness.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
