wanglijie95 commented on code in PR #20345:
URL: https://github.com/apache/flink/pull/20345#discussion_r928632978
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java:
##########
@@ -109,14 +132,35 @@ void testSpeculativeExecutionWithAdaptiveParallelism()
throws Exception {
void testBlockSlowNodeInSpeculativeExecution() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.BLOCK_SLOW_NODE_DURATION,
Duration.ofMinutes(1));
- JobClient client = executeJobAsync(configuration);
+ JobClient client = executeJobAsync(configuration,
this::setupJobWithSlowMap);
assertThatThrownBy(
() -> client.getJobExecutionResult().get(30,
TimeUnit.SECONDS),
"The local node is expected to be blocked but it is
not.")
.isInstanceOf(TimeoutException.class);
}
+ @Test
+ public void testSpeculativeExecutionOfSourceFunction() throws Exception {
Review Comment:
public is not needed
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java:
##########
@@ -109,14 +132,35 @@ void testSpeculativeExecutionWithAdaptiveParallelism()
throws Exception {
void testBlockSlowNodeInSpeculativeExecution() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.BLOCK_SLOW_NODE_DURATION,
Duration.ofMinutes(1));
- JobClient client = executeJobAsync(configuration);
+ JobClient client = executeJobAsync(configuration,
this::setupJobWithSlowMap);
assertThatThrownBy(
() -> client.getJobExecutionResult().get(30,
TimeUnit.SECONDS),
"The local node is expected to be blocked but it is
not.")
.isInstanceOf(TimeoutException.class);
}
+ @Test
+ public void testSpeculativeExecutionOfSourceFunction() throws Exception {
+ executeJob(this::setupJobWithSlowSourceFunction);
+ waitUntilJobArchived();
+ checkResults();
+ }
+
+ @Test
+ public void testSpeculativeExecutionOfInputFormatSource() throws Exception
{
Review Comment:
public is not needed
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java:
##########
@@ -109,14 +132,35 @@ void testSpeculativeExecutionWithAdaptiveParallelism()
throws Exception {
void testBlockSlowNodeInSpeculativeExecution() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.BLOCK_SLOW_NODE_DURATION,
Duration.ofMinutes(1));
- JobClient client = executeJobAsync(configuration);
+ JobClient client = executeJobAsync(configuration,
this::setupJobWithSlowMap);
assertThatThrownBy(
() -> client.getJobExecutionResult().get(30,
TimeUnit.SECONDS),
"The local node is expected to be blocked but it is
not.")
.isInstanceOf(TimeoutException.class);
}
+ @Test
+ public void testSpeculativeExecutionOfSourceFunction() throws Exception {
+ executeJob(this::setupJobWithSlowSourceFunction);
+ waitUntilJobArchived();
+ checkResults();
+ }
+
+ @Test
+ public void testSpeculativeExecutionOfInputFormatSource() throws Exception
{
+ executeJob(this::setupJobWithSlowInputFormatSource);
+ waitUntilJobArchived();
+ checkResults();
+ }
+
+ @Test
+ public void testSpeculativeExecutionOfNewSource() throws Exception {
Review Comment:
public is not needed
--
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]
