You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/09/03 13:53:36 UTC
[flink] branch release-1.14 updated (94387a4 -> 2e76974)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 94387a4 [FLINK-24021][HA]Handle curator framework start error by register UnhandledErrorListener before start zk client
new e24ebea [FLINK-23916][python] Add python API to set tolerable checkpoint failure number
new 2e76974 [FLINK-23916][docs] Update the documentation how to set tolerable checkpoint failure number
The 2 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:
.../datastream/fault-tolerance/checkpointing.md | 24 +++++++++++++-----
.../pyflink/datastream/checkpoint_config.py | 29 ++++++++++++++++++++++
.../datastream/tests/test_check_point_config.py | 8 ++++++
.../api/environment/CheckpointConfig.java | 9 ++++---
4 files changed, 60 insertions(+), 10 deletions(-)
[flink] 01/02: [FLINK-23916][python] Add python API to set
tolerable checkpoint failure number
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e24ebeac78afc6df9842b3ee6419af1c752d603b
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Sep 1 15:54:32 2021 +0200
[FLINK-23916][python] Add python API to set tolerable checkpoint failure number
---
.../pyflink/datastream/checkpoint_config.py | 29 ++++++++++++++++++++++
.../datastream/tests/test_check_point_config.py | 8 ++++++
2 files changed, 37 insertions(+)
diff --git a/flink-python/pyflink/datastream/checkpoint_config.py b/flink-python/pyflink/datastream/checkpoint_config.py
index 8ab2918..4119a91 100644
--- a/flink-python/pyflink/datastream/checkpoint_config.py
+++ b/flink-python/pyflink/datastream/checkpoint_config.py
@@ -214,6 +214,35 @@ class CheckpointConfig(object):
self._j_checkpoint_config.setFailOnCheckpointingErrors(fail_on_checkpointing_errors)
return self
+ def get_tolerable_checkpoint_failure_number(self) -> int:
+ """
+ Get the defined number of consecutive checkpoint failures that will be tolerated, before the
+ whole job is failed over.
+
+ :return: The maximum number of tolerated checkpoint failures.
+ """
+ return self._j_checkpoint_config.getTolerableCheckpointFailureNumber()
+
+ def set_tolerable_checkpoint_failure_number(self,
+ tolerable_checkpoint_failure_number: int
+ ) -> 'CheckpointConfig':
+ """
+ This defines how many consecutive checkpoint failures will be tolerated, before the whole
+ job is failed over. The default value is `0`, which means no checkpoint failures will be
+ tolerated, and the job will fail on first reported checkpoint failure.
+
+ Example:
+ ::
+
+ >>> config.set_tolerable_checkpoint_failure_number(2)
+
+ :param tolerable_checkpoint_failure_number: The maximum number of tolerated checkpoint
+ failures.
+ """
+ self._j_checkpoint_config.setTolerableCheckpointFailureNumber(
+ tolerable_checkpoint_failure_number)
+ return self
+
def enable_externalized_checkpoints(
self,
cleanup_mode: 'ExternalizedCheckpointCleanup') -> 'CheckpointConfig':
diff --git a/flink-python/pyflink/datastream/tests/test_check_point_config.py b/flink-python/pyflink/datastream/tests/test_check_point_config.py
index d4fcb9c..11420f1 100644
--- a/flink-python/pyflink/datastream/tests/test_check_point_config.py
+++ b/flink-python/pyflink/datastream/tests/test_check_point_config.py
@@ -109,6 +109,14 @@ class CheckpointConfigTests(PyFlinkTestCase):
self.assertFalse(self.checkpoint_config.is_fail_on_checkpointing_errors())
+ def test_get_set_tolerable_checkpoint_failure_number(self):
+
+ self.assertEqual(self.checkpoint_config.get_tolerable_checkpoint_failure_number(), 0)
+
+ self.checkpoint_config.set_tolerable_checkpoint_failure_number(2)
+
+ self.assertEqual(self.checkpoint_config.get_tolerable_checkpoint_failure_number(), 2)
+
def test_get_set_externalized_checkpoints_cleanup(self):
self.assertFalse(self.checkpoint_config.is_externalized_checkpoints_enabled())
[flink] 02/02: [FLINK-23916][docs] Update the documentation how to
set tolerable checkpoint failure number
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2e769746bf29f06d62568a88f5a3e919bc568cbb
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Sep 1 15:34:43 2021 +0200
[FLINK-23916][docs] Update the documentation how to set tolerable checkpoint failure number
---
.../datastream/fault-tolerance/checkpointing.md | 24 ++++++++++++++++------
.../api/environment/CheckpointConfig.java | 9 ++++----
2 files changed, 23 insertions(+), 10 deletions(-)
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
index 2a8d96a..6809e85 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
@@ -67,6 +67,10 @@ Other parameters for checkpointing include:
Note that this value also implies that the number of concurrent checkpoints is *one*.
+ - *tolerable checkpoint failure number*: This defines how many consecutive checkpoint failures will be tolerated,
+ before the whole job is failed over. The default value is `0`, which means no checkpoint failures will be tolerated,
+ and the job will fail on first reported checkpoint failure.
+
- *number of concurrent checkpoints*: By default, the system will not trigger another checkpoint while one is still in progress.
This ensures that the topology does not spend too much time on checkpoints and not make progress with processing the streams.
It is possible to allow for multiple overlapping checkpoints, which is interesting for pipelines that have a certain processing delay
@@ -77,8 +81,6 @@ Other parameters for checkpointing include:
- *externalized checkpoints*: You can configure periodic checkpoints to be persisted externally. Externalized checkpoints write their meta data out to persistent storage and are *not* automatically cleaned up when the job fails. This way, you will have a checkpoint around to resume from if your job fails. There are more details in the [deployment notes on externalized checkpoints]({{< ref "docs/ops/state/checkpoints" >}}#externalized-checkpoints).
- - *fail/continue task on checkpoint errors*: This determines if a task will be failed if an error occurs in the execution of the task's checkpoint procedure. This is the default behaviour. Alternatively, when this is disabled, the task will simply decline the checkpoint to the checkpoint coordinator and continue running.
-
- *unaligned checkpoints*: You can enable [unaligned checkpoints]({{< ref "docs/ops/state/unaligned_checkpoints" >}}) to greatly reduce checkpointing times under backpressure. Only works for exactly-once checkpoints and with number of concurrent checkpoints of 1.
{{< tabs "4b9c6a74-8a45-4ad2-9e80-52fe44a85991" >}}
@@ -100,10 +102,13 @@ env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig().setCheckpointTimeout(60000);
+// only two consecutive checkpoint failures are tolerated
+env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
+
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
-// enable externalized checkpoints which are retained
+// enable externalized checkpoints which are retained
// after job cancellation
env.getCheckpointConfig().enableExternalizedCheckpoints(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
@@ -133,13 +138,17 @@ env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig.setCheckpointTimeout(60000)
-// prevent the tasks from failing if an error happens in their checkpointing,
-// the checkpoint will just be declined.
-env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)
+// only two consecutive checkpoint failures are tolerated
+env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2)
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
+// enable externalized checkpoints which are retained
+// after job cancellation
+env.getCheckpointConfig().enableExternalizedCheckpoints(
+ ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
+
// enables the experimental unaligned checkpoints
env.getCheckpointConfig.enableUnalignedCheckpoints()
@@ -165,6 +174,9 @@ env.get_checkpoint_config().set_min_pause_between_checkpoints(500)
# checkpoints have to complete within one minute, or are discarded
env.get_checkpoint_config().set_checkpoint_timeout(60000)
+# only two consecutive checkpoint failures are tolerated
+env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(2)
+
# allow only one checkpoint to be in progress at the same time
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
index 8204c67..bdede360 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
@@ -399,8 +399,8 @@ public class CheckpointConfig implements java.io.Serializable {
}
/**
- * Get the tolerable checkpoint failure number which used by the checkpoint failure manager to
- * determine when we need to fail the job.
+ * Get the defined number of consecutive checkpoint failures that will be tolerated, before the
+ * whole job is failed over.
*
* <p>If the {@link #tolerableCheckpointFailureNumber} has not been configured, this method
* would return 0 which means the checkpoint failure manager would not tolerate any declined
@@ -414,8 +414,9 @@ public class CheckpointConfig implements java.io.Serializable {
}
/**
- * Set the tolerable checkpoint failure number, the default value is 0 that means we do not
- * tolerance any checkpoint failure.
+ * This defines how many consecutive checkpoint failures will be tolerated, before the whole job
+ * is failed over. The default value is `0`, which means no checkpoint failures will be
+ * tolerated, and the job will fail on first reported checkpoint failure.
*/
public void setTolerableCheckpointFailureNumber(int tolerableCheckpointFailureNumber) {
if (tolerableCheckpointFailureNumber < 0) {