You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2019/04/16 09:43:34 UTC
[flink] branch master updated (6163af6 -> 3e5a240)
This is an automated email from the ASF dual-hosted git repository.
gary pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 6163af6 [hotfix] [typo] fix typo in debugging_event_time
new d27efa5 [FLINK-12193][tests] Add option to disable failOnCheckpointingErrors
new 3e36be7 [hotfix][tests] Split DataStreamAllroundTestJobFactory#setupEnvironment into smaller methods
new a782169 [hotfix][tests] Reorder ConfigOptions in DataStreamAllroundTestJobFactory
new 3e5a240 [hotfix][tests] Fix typo in exception message
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../tests/DataStreamAllroundTestJobFactory.java | 101 ++++++++++++---------
1 file changed, 59 insertions(+), 42 deletions(-)
[flink] 04/04: [hotfix][tests] Fix typo in exception message
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3e5a240a35936f4b953bbd3226dfa81d9ed3ab43
Author: Gary Yao <ga...@apache.org>
AuthorDate: Tue Apr 16 11:29:37 2019 +0200
[hotfix][tests] Fix typo in exception message
---
.../apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
index a897bf2..913d030 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
@@ -303,7 +303,7 @@ public class DataStreamAllroundTestJobFactory {
restartStrategy = RestartStrategies.noRestart();
break;
default:
- throw new IllegalArgumentException("Unkown restart strategy: " + restartStrategyConfig);
+ throw new IllegalArgumentException("Unknown restart strategy: " + restartStrategyConfig);
}
env.setRestartStrategy(restartStrategy);
}
[flink] 01/04: [FLINK-12193][tests] Add option to disable
failOnCheckpointingErrors
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d27efa54a1bca4b44d24b2a31b0fa1a31a705923
Author: Gary Yao <ga...@apache.org>
AuthorDate: Mon Apr 15 12:55:16 2019 +0200
[FLINK-12193][tests] Add option to disable failOnCheckpointingErrors
Add option to disable failOnCheckpointingErrors in DataStreamAllroundTestJob.
---
.../streaming/tests/DataStreamAllroundTestJobFactory.java | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
index af39522..b9869f3 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
@@ -80,6 +80,7 @@ import static org.apache.flink.streaming.tests.TestOperatorEnum.RESULT_TYPE_QUER
* <li>environment.checkpoint_interval (long, default - 1000): the checkpoint interval.</li>
* <li>environment.externalize_checkpoint (boolean, default - false): whether or not checkpoints should be externalized.</li>
* <li>environment.externalize_checkpoint.cleanup (String, default - 'retain'): Configures the cleanup mode for externalized checkpoints. Can be 'retain' or 'delete'.</li>
+ * <li>environment.fail_on_checkpointing_errors (String, default - true): Sets the expected behaviour for tasks in case that they encounter an error in their checkpointing procedure.</li>
* <li>environment.parallelism (int, default - 1): parallelism to use for the job.</li>
* <li>environment.max_parallelism (int, default - 128): max parallelism to use for the job</li>
* <li>environment.restart_strategy (String, default - 'fixed_delay'): The failure restart strategy to use. Can be 'fixed_delay' or 'no_restart'.</li>
@@ -168,6 +169,10 @@ public class DataStreamAllroundTestJobFactory {
.key("environment.externalize_checkpoint.cleanup")
.defaultValue("retain");
+ private static final ConfigOption<Boolean> ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS = ConfigOptions
+ .key("environment.fail_on_checkpointing_errors")
+ .defaultValue(true);
+
private static final ConfigOption<String> STATE_BACKEND = ConfigOptions
.key("state_backend")
.defaultValue("file")
@@ -308,10 +313,14 @@ public class DataStreamAllroundTestJobFactory {
default:
throw new IllegalArgumentException("Unknown clean up mode for externalized checkpoints: " + cleanupModeConfig);
}
-
env.getCheckpointConfig().enableExternalizedCheckpoints(cleanupMode);
}
+ final boolean failOnCheckpointingErrors = pt.getBoolean(
+ ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.key(),
+ ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.defaultValue());
+ env.getCheckpointConfig().setFailOnCheckpointingErrors(failOnCheckpointingErrors);
+
// make parameters available in the web interface
env.getConfig().setGlobalJobParameters(pt);
}
[flink] 02/04: [hotfix][tests] Split
DataStreamAllroundTestJobFactory#setupEnvironment into smaller methods
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3e36be7ab936ad4b293966dcbacd0f43e4a747db
Author: Gary Yao <ga...@apache.org>
AuthorDate: Mon Apr 15 13:38:21 2019 +0200
[hotfix][tests] Split DataStreamAllroundTestJobFactory#setupEnvironment into smaller methods
---
.../tests/DataStreamAllroundTestJobFactory.java | 82 ++++++++++++----------
1 file changed, 45 insertions(+), 37 deletions(-)
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
index b9869f3..1319232 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
@@ -53,6 +53,7 @@ import org.apache.flink.streaming.tests.artificialstate.builder.ArtificialListSt
import org.apache.flink.streaming.tests.artificialstate.builder.ArtificialStateBuilder;
import org.apache.flink.streaming.tests.artificialstate.builder.ArtificialValueStateBuilder;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -230,8 +231,18 @@ public class DataStreamAllroundTestJobFactory {
.defaultValue(250L);
public static void setupEnvironment(StreamExecutionEnvironment env, ParameterTool pt) throws Exception {
+ setupCheckpointing(env, pt);
+ setupParallelism(env, pt);
+ setupRestartStrategy(env, pt);
+ setupStateBackend(env, pt);
- // set checkpointing semantics
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(pt);
+ }
+
+ private static void setupCheckpointing(final StreamExecutionEnvironment env, final ParameterTool pt) {
String semantics = pt.get(TEST_SEMANTICS.key(), TEST_SEMANTICS.defaultValue());
long checkpointInterval = pt.getLong(ENVIRONMENT_CHECKPOINT_INTERVAL.key(), ENVIRONMENT_CHECKPOINT_INTERVAL.defaultValue());
CheckpointingMode checkpointingMode = semantics.equalsIgnoreCase("exactly-once")
@@ -240,14 +251,41 @@ public class DataStreamAllroundTestJobFactory {
env.enableCheckpointing(checkpointInterval, checkpointingMode);
- // use event time
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ boolean enableExternalizedCheckpoints = pt.getBoolean(
+ ENVIRONMENT_EXTERNALIZE_CHECKPOINT.key(),
+ ENVIRONMENT_EXTERNALIZE_CHECKPOINT.defaultValue());
+
+ if (enableExternalizedCheckpoints) {
+ String cleanupModeConfig = pt.get(
+ ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP.key(),
+ ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP.defaultValue());
+
+ CheckpointConfig.ExternalizedCheckpointCleanup cleanupMode;
+ switch (cleanupModeConfig) {
+ case "retain":
+ cleanupMode = CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;
+ break;
+ case "delete":
+ cleanupMode = CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION;
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown clean up mode for externalized checkpoints: " + cleanupModeConfig);
+ }
+ env.getCheckpointConfig().enableExternalizedCheckpoints(cleanupMode);
+ }
+
+ final boolean failOnCheckpointingErrors = pt.getBoolean(
+ ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.key(),
+ ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.defaultValue());
+ env.getCheckpointConfig().setFailOnCheckpointingErrors(failOnCheckpointingErrors);
+ }
- // parallelism
+ private static void setupParallelism(final StreamExecutionEnvironment env, final ParameterTool pt) {
env.setParallelism(pt.getInt(ENVIRONMENT_PARALLELISM.key(), ENVIRONMENT_PARALLELISM.defaultValue()));
env.setMaxParallelism(pt.getInt(ENVIRONMENT_MAX_PARALLELISM.key(), ENVIRONMENT_MAX_PARALLELISM.defaultValue()));
+ }
- // restart strategy
+ private static void setupRestartStrategy(final StreamExecutionEnvironment env, final ParameterTool pt) {
String restartStrategyConfig = pt.get(ENVIRONMENT_RESTART_STRATEGY.key());
if (restartStrategyConfig != null) {
RestartStrategies.RestartStrategyConfiguration restartStrategy;
@@ -269,8 +307,9 @@ public class DataStreamAllroundTestJobFactory {
}
env.setRestartStrategy(restartStrategy);
}
+ }
- // state backend
+ private static void setupStateBackend(final StreamExecutionEnvironment env, final ParameterTool pt) throws IOException {
final String stateBackend = pt.get(
STATE_BACKEND.key(),
STATE_BACKEND.defaultValue());
@@ -292,37 +331,6 @@ public class DataStreamAllroundTestJobFactory {
} else {
throw new IllegalArgumentException("Unknown backend requested: " + stateBackend);
}
-
- boolean enableExternalizedCheckpoints = pt.getBoolean(
- ENVIRONMENT_EXTERNALIZE_CHECKPOINT.key(),
- ENVIRONMENT_EXTERNALIZE_CHECKPOINT.defaultValue());
-
- if (enableExternalizedCheckpoints) {
- String cleanupModeConfig = pt.get(
- ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP.key(),
- ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP.defaultValue());
-
- CheckpointConfig.ExternalizedCheckpointCleanup cleanupMode;
- switch (cleanupModeConfig) {
- case "retain":
- cleanupMode = CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;
- break;
- case "delete":
- cleanupMode = CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION;
- break;
- default:
- throw new IllegalArgumentException("Unknown clean up mode for externalized checkpoints: " + cleanupModeConfig);
- }
- env.getCheckpointConfig().enableExternalizedCheckpoints(cleanupMode);
- }
-
- final boolean failOnCheckpointingErrors = pt.getBoolean(
- ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.key(),
- ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS.defaultValue());
- env.getCheckpointConfig().setFailOnCheckpointingErrors(failOnCheckpointingErrors);
-
- // make parameters available in the web interface
- env.getConfig().setGlobalJobParameters(pt);
}
static SourceFunction<Event> createEventSource(ParameterTool pt) {
[flink] 03/04: [hotfix][tests] Reorder ConfigOptions in
DataStreamAllroundTestJobFactory
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a782169f422e7b1ac36dd74dd8b6aa2c090f8756
Author: Gary Yao <ga...@apache.org>
AuthorDate: Mon Apr 15 16:52:55 2019 +0200
[hotfix][tests] Reorder ConfigOptions in DataStreamAllroundTestJobFactory
Reorder so that checkpoint-related ConfigOptions appear consecutively.
---
.../tests/DataStreamAllroundTestJobFactory.java | 24 +++++++++++-----------
1 file changed, 12 insertions(+), 12 deletions(-)
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
index 1319232..a897bf2 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
@@ -142,6 +142,18 @@ public class DataStreamAllroundTestJobFactory {
.key("environment.checkpoint_interval")
.defaultValue(1000L);
+ private static final ConfigOption<Boolean> ENVIRONMENT_EXTERNALIZE_CHECKPOINT = ConfigOptions
+ .key("environment.externalize_checkpoint")
+ .defaultValue(false);
+
+ private static final ConfigOption<String> ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP = ConfigOptions
+ .key("environment.externalize_checkpoint.cleanup")
+ .defaultValue("retain");
+
+ private static final ConfigOption<Boolean> ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS = ConfigOptions
+ .key("environment.fail_on_checkpointing_errors")
+ .defaultValue(true);
+
private static final ConfigOption<Integer> ENVIRONMENT_PARALLELISM = ConfigOptions
.key("environment.parallelism")
.defaultValue(1);
@@ -162,18 +174,6 @@ public class DataStreamAllroundTestJobFactory {
.key("environment.restart_strategy.fixed.delay")
.defaultValue(0L);
- private static final ConfigOption<Boolean> ENVIRONMENT_EXTERNALIZE_CHECKPOINT = ConfigOptions
- .key("environment.externalize_checkpoint")
- .defaultValue(false);
-
- private static final ConfigOption<String> ENVIRONMENT_EXTERNALIZE_CHECKPOINT_CLEANUP = ConfigOptions
- .key("environment.externalize_checkpoint.cleanup")
- .defaultValue("retain");
-
- private static final ConfigOption<Boolean> ENVIRONMENT_FAIL_ON_CHECKPOINTING_ERRORS = ConfigOptions
- .key("environment.fail_on_checkpointing_errors")
- .defaultValue(true);
-
private static final ConfigOption<String> STATE_BACKEND = ConfigOptions
.key("state_backend")
.defaultValue("file")