You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/07/25 09:00:19 UTC
[GitHub] [flink] wanglijie95 commented on a diff in pull request #20345: [FLINK-28610][runtime] Enable speculative execution for sources
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org