You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2024/09/24 14:06:33 UTC
(flink) 02/08: [FLINK-36274][Checkpoint] Remove deprecated `ExternalizedCheckpointCleanup`
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5d463de174a8f7e5df79ff5734f7b75639671c95
Author: Zakelly <za...@gmail.com>
AuthorDate: Wed Sep 18 12:08:21 2024 +0800
[FLINK-36274][Checkpoint] Remove deprecated `ExternalizedCheckpointCleanup`
---
.../datastream/fault-tolerance/checkpointing.md | 2 +-
.../datastream/fault-tolerance/checkpointing.md | 2 +-
.../reference/pyflink.datastream/checkpoint.rst | 4 -
flink-python/pyflink/datastream/__init__.py | 3 +-
.../pyflink/datastream/checkpoint_config.py | 152 +--------------------
.../externalized_checkpoint_retention.py | 3 +-
.../datastream/tests/test_check_point_config.py | 26 +---
.../api/environment/CheckpointConfig.java | 142 +------------------
.../environment/ExecutionCheckpointingOptions.java | 35 -----
.../CheckpointConfigFromConfigurationTest.java | 38 ------
10 files changed, 10 insertions(+), 397 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/checkpointing.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/checkpointing.md
index 3d01e0ffddf..83ade4fbf68 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/checkpointing.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/checkpointing.md
@@ -171,7 +171,7 @@ env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(2)
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
# 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
-env.get_checkpoint_config().enable_externalized_checkpoints(
+env.get_checkpoint_config().set_externalized_checkpoint_retention(
ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
# 开启实验性的 unaligned checkpoints
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
index ad68b3ecc23..532fc618e72 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
@@ -200,7 +200,7 @@ env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(2)
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
# enable externalized checkpoints which are retained after job cancellation
-env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
+env.get_checkpoint_config().set_externalized_checkpoint_retention(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
# enables the unaligned checkpoints
env.get_checkpoint_config().enable_unaligned_checkpoints()
diff --git a/flink-python/docs/reference/pyflink.datastream/checkpoint.rst b/flink-python/docs/reference/pyflink.datastream/checkpoint.rst
index d3321b774c7..37c7e418880 100644
--- a/flink-python/docs/reference/pyflink.datastream/checkpoint.rst
+++ b/flink-python/docs/reference/pyflink.datastream/checkpoint.rst
@@ -67,10 +67,7 @@ The default limit of concurrently happening checkpoints: one.
CheckpointConfig.set_fail_on_checkpointing_errors
CheckpointConfig.get_tolerable_checkpoint_failure_number
CheckpointConfig.set_tolerable_checkpoint_failure_number
- CheckpointConfig.enable_externalized_checkpoints
- CheckpointConfig.set_externalized_checkpoint_cleanup
CheckpointConfig.is_externalized_checkpoints_enabled
- CheckpointConfig.get_externalized_checkpoint_cleanup
CheckpointConfig.is_unaligned_checkpoints_enabled
CheckpointConfig.enable_unaligned_checkpoints
CheckpointConfig.disable_unaligned_checkpoints
@@ -78,7 +75,6 @@ The default limit of concurrently happening checkpoints: one.
CheckpointConfig.get_alignment_timeout
CheckpointConfig.set_force_unaligned_checkpoints
CheckpointConfig.is_force_unaligned_checkpoints
- ExternalizedCheckpointCleanup
ExternalizedCheckpointRetention
diff --git a/flink-python/pyflink/datastream/__init__.py b/flink-python/pyflink/datastream/__init__.py
index dbb5b978a01..ce543694dc0 100644
--- a/flink-python/pyflink/datastream/__init__.py
+++ b/flink-python/pyflink/datastream/__init__.py
@@ -259,7 +259,7 @@ Other important classes:
- :class:`OutputTag`:
Tag with a name and type for identifying side output of an operator
"""
-from pyflink.datastream.checkpoint_config import CheckpointConfig, ExternalizedCheckpointCleanup
+from pyflink.datastream.checkpoint_config import CheckpointConfig
from pyflink.datastream.externalized_checkpoint_retention import ExternalizedCheckpointRetention
from pyflink.datastream.checkpointing_mode import CheckpointingMode
from pyflink.datastream.data_stream import DataStream, KeyedStream, WindowedStream, \
@@ -317,7 +317,6 @@ __all__ = [
'TimerService',
'CheckpointingMode',
'CheckpointConfig',
- 'ExternalizedCheckpointCleanup',
'ExternalizedCheckpointRetention',
'StateBackend',
'HashMapStateBackend',
diff --git a/flink-python/pyflink/datastream/checkpoint_config.py b/flink-python/pyflink/datastream/checkpoint_config.py
index d5b2e7fa0f5..461d91fe455 100644
--- a/flink-python/pyflink/datastream/checkpoint_config.py
+++ b/flink-python/pyflink/datastream/checkpoint_config.py
@@ -15,15 +15,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-from enum import Enum
from typing import Optional
from pyflink.common import Duration
from pyflink.datastream.externalized_checkpoint_retention import ExternalizedCheckpointRetention
from pyflink.datastream.checkpointing_mode import CheckpointingMode
-from pyflink.java_gateway import get_gateway
-__all__ = ['CheckpointConfig', 'ExternalizedCheckpointCleanup']
+__all__ = ['CheckpointConfig']
class CheckpointConfig(object):
@@ -243,82 +241,6 @@ class CheckpointConfig(object):
tolerable_checkpoint_failure_number)
return self
- def enable_externalized_checkpoints(
- self,
- cleanup_mode: 'ExternalizedCheckpointCleanup') -> 'CheckpointConfig':
- """
- Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled
- automatically unless the mode is set to
- :data:`ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS`.
-
- Externalized checkpoints write their meta data out to persistent storage and are **not**
- automatically cleaned up when the owning job fails or is suspended (terminating with job
- status ``FAILED`` or ``SUSPENDED``). In this case, you have to manually clean up the
- checkpoint state, both the meta data and actual program state.
-
- The :class:`ExternalizedCheckpointCleanup` mode defines how an externalized checkpoint
- should be cleaned up on job cancellation. If you choose to retain externalized checkpoints
- on cancellation you have to handle checkpoint clean-up manually when you cancel the job as
- well (terminating with job status ``CANCELED``).
-
- The target directory for externalized checkpoints is configured via
- ``org.apache.flink.configuration.CheckpointingOptions#CHECKPOINTS_DIRECTORY``.
-
- Example:
- ::
-
- >>> config.enable_externalized_checkpoints(
- ... ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-
- :param cleanup_mode: Externalized checkpoint clean-up behaviour, the mode could be
- :data:`ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION`,
- :data:`ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION` or
- :data:`ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS`
-
- .. note:: Deprecated in 1.15. Use :func:`set_externalized_checkpoint_cleanup` instead.
- """
- self._j_checkpoint_config.enableExternalizedCheckpoints(
- ExternalizedCheckpointCleanup._to_j_externalized_checkpoint_cleanup(cleanup_mode))
- return self
-
- def set_externalized_checkpoint_cleanup(
- self,
- cleanup_mode: 'ExternalizedCheckpointCleanup') -> 'CheckpointConfig':
- """
- Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled
- automatically unless the mode is set to
- :data:`ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS`.
-
- Externalized checkpoints write their meta data out to persistent storage and are **not**
- automatically cleaned up when the owning job fails or is suspended (terminating with job
- status ``FAILED`` or ``SUSPENDED``). In this case, you have to manually clean up the
- checkpoint state, both the meta data and actual program state.
-
- The :class:`ExternalizedCheckpointCleanup` mode defines how an externalized checkpoint
- should be cleaned up on job cancellation. If you choose to retain externalized checkpoints
- on cancellation you have to handle checkpoint clean-up manually when you cancel the job as
- well (terminating with job status ``CANCELED``).
-
- The target directory for externalized checkpoints is configured via
- ``org.apache.flink.configuration.CheckpointingOptions#CHECKPOINTS_DIRECTORY``.
-
- Example:
- ::
-
- >>> config.set_externalized_checkpoint_cleanup(
- ... ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-
- :param cleanup_mode: Externalized checkpoint clean-up behaviour, the mode could be
- :data:`ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION`,
- :data:`ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION` or
- :data:`ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS`
-
- note:: Deprecated. Use :func:`set_externalized_checkpoint_retention` instead.
- """
- self._j_checkpoint_config.setExternalizedCheckpointCleanup(
- ExternalizedCheckpointCleanup._to_j_externalized_checkpoint_cleanup(cleanup_mode))
- return self
-
def set_externalized_checkpoint_retention(
self,
retention_mode: 'ExternalizedCheckpointRetention') -> 'CheckpointConfig':
@@ -363,22 +285,6 @@ class CheckpointConfig(object):
"""
return self._j_checkpoint_config.isExternalizedCheckpointsEnabled()
- def get_externalized_checkpoint_cleanup(self) -> Optional['ExternalizedCheckpointCleanup']:
- """
- Returns the cleanup behaviour for externalized checkpoints.
-
- :return: The cleanup behaviour for externalized checkpoints or ``None`` if none is
- configured.
-
- note:: Deprecated. Use :func:`get_externalized_checkpoint_retention` instead.
- """
- cleanup_mode = self._j_checkpoint_config.getExternalizedCheckpointCleanup()
- if cleanup_mode is None:
- return None
- else:
- return ExternalizedCheckpointCleanup._from_j_externalized_checkpoint_cleanup(
- cleanup_mode)
-
def get_externalized_checkpoint_retention(self) -> Optional['ExternalizedCheckpointRetention']:
"""
Returns the cleanup behaviour for externalized checkpoints.
@@ -452,7 +358,7 @@ class CheckpointConfig(object):
def get_alignment_timeout(self) -> 'Duration':
"""
Returns the alignment timeout, as configured via :func:`set_alignment_timeout` or
- ``org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions#ALIGNMENT_TIMEOUT``.
+ ``org.apache.flink.configuration.CheckpointingOptions#ALIGNED_CHECKPOINT_TIMEOUT``.
:return: the alignment timeout.
"""
@@ -478,57 +384,3 @@ class CheckpointConfig(object):
:return: True, if unaligned checkpoints are forced, false otherwise.
"""
return self._j_checkpoint_config.isForceUnalignedCheckpoints()
-
-
-class ExternalizedCheckpointCleanup(Enum):
- """
- Cleanup behaviour for externalized checkpoints when the job is cancelled.
-
- :data:`DELETE_ON_CANCELLATION`:
-
- Delete externalized checkpoints on job cancellation.
-
- All checkpoint state will be deleted when you cancel the owning
- job, both the meta data and actual program state. Therefore, you
- cannot resume from externalized checkpoints after the job has been
- cancelled.
-
- Note that checkpoint state is always kept if the job terminates
- with state ``FAILED``.
-
- :data:`RETAIN_ON_CANCELLATION`:
-
- Retain externalized checkpoints on job cancellation.
-
- All checkpoint state is kept when you cancel the owning job. You
- have to manually delete both the checkpoint meta data and actual
- program state after cancelling the job.
-
- Note that checkpoint state is always kept if the job terminates
- with state ``FAILED``.
-
- :data:`NO_EXTERNALIZED_CHECKPOINTS`:
-
- Externalized checkpoints are disabled completely.
-
- note:: Deprecated. Please use
- pyflink.datastream.externalized_checkpoint_retention.ExternalizedCheckpointRetention instead.
- """
-
- DELETE_ON_CANCELLATION = 0
-
- RETAIN_ON_CANCELLATION = 1
-
- NO_EXTERNALIZED_CHECKPOINTS = 2
-
- @staticmethod
- def _from_j_externalized_checkpoint_cleanup(j_cleanup_mode) \
- -> 'ExternalizedCheckpointCleanup':
- return ExternalizedCheckpointCleanup[j_cleanup_mode.name()]
-
- def _to_j_externalized_checkpoint_cleanup(self):
- gateway = get_gateway()
- JExternalizedCheckpointCleanup = \
- gateway.jvm.org.apache.flink.streaming.api.environment.CheckpointConfig \
- .ExternalizedCheckpointCleanup
- return getattr(JExternalizedCheckpointCleanup, self.name)
diff --git a/flink-python/pyflink/datastream/externalized_checkpoint_retention.py b/flink-python/pyflink/datastream/externalized_checkpoint_retention.py
index f66aad517c0..3857d5d27fa 100644
--- a/flink-python/pyflink/datastream/externalized_checkpoint_retention.py
+++ b/flink-python/pyflink/datastream/externalized_checkpoint_retention.py
@@ -66,6 +66,5 @@ class ExternalizedCheckpointRetention(Enum):
def _to_j_externalized_checkpoint_retention(self):
gateway = get_gateway()
JExternalizedCheckpointRetention = \
- gateway.jvm.org.apache.flink.streaming.api.environment.CheckpointConfig \
- .ExternalizedCheckpointRetention
+ gateway.jvm.org.apache.flink.configuration.ExternalizedCheckpointRetention
return getattr(JExternalizedCheckpointRetention, self.name)
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 9c798515164..382a374eb97 100644
--- a/flink-python/pyflink/datastream/tests/test_check_point_config.py
+++ b/flink-python/pyflink/datastream/tests/test_check_point_config.py
@@ -18,7 +18,6 @@
from pyflink.common import Duration
from pyflink.datastream import (CheckpointConfig,
CheckpointingMode,
- ExternalizedCheckpointCleanup,
ExternalizedCheckpointRetention,
StreamExecutionEnvironment)
from pyflink.java_gateway import get_gateway
@@ -120,27 +119,6 @@ class CheckpointConfigTests(PyFlinkTestCase):
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())
-
- self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_cleanup(),
- ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS)
-
- self.checkpoint_config.enable_externalized_checkpoints(
- ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-
- self.assertTrue(self.checkpoint_config.is_externalized_checkpoints_enabled())
-
- self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_cleanup(),
- ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-
- self.checkpoint_config.enable_externalized_checkpoints(
- ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)
-
- self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_cleanup(),
- ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)
-
def test_get_set_externalized_checkpoints_retention(self):
self.assertFalse(self.checkpoint_config.is_externalized_checkpoints_enabled())
@@ -148,7 +126,7 @@ class CheckpointConfigTests(PyFlinkTestCase):
self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_retention(),
ExternalizedCheckpointRetention.NO_EXTERNALIZED_CHECKPOINTS)
- self.checkpoint_config.enable_externalized_checkpoints(
+ self.checkpoint_config.set_externalized_checkpoint_retention(
ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
self.assertTrue(self.checkpoint_config.is_externalized_checkpoints_enabled())
@@ -156,7 +134,7 @@ class CheckpointConfigTests(PyFlinkTestCase):
self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_retention(),
ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
- self.checkpoint_config.enable_externalized_checkpoints(
+ self.checkpoint_config.set_externalized_checkpoint_retention(
ExternalizedCheckpointRetention.DELETE_ON_CANCELLATION)
self.assertEqual(self.checkpoint_config.get_externalized_checkpoint_retention(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
index 0448fab868f..5b09d8a3d6c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
@@ -25,11 +25,9 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.ExternalizedCheckpointRetention;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.configuration.StateRecoveryOptions;
-import org.apache.flink.configuration.description.InlineElement;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.util.Preconditions;
@@ -39,7 +37,6 @@ import org.slf4j.LoggerFactory;
import java.time.Duration;
-import static org.apache.flink.configuration.description.TextElement.text;
import static org.apache.flink.runtime.checkpoint.CheckpointFailureManager.UNLIMITED_TOLERABLE_FAILURE_NUMBER;
import static org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration.MINIMAL_CHECKPOINT_TIME;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -474,34 +471,6 @@ public class CheckpointConfig implements java.io.Serializable {
CheckpointingOptions.TOLERABLE_FAILURE_NUMBER, tolerableCheckpointFailureNumber);
}
- /**
- * Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled
- * automatically unless the mode is set to {@link
- * ExternalizedCheckpointCleanup#NO_EXTERNALIZED_CHECKPOINTS}.
- *
- * <p>Externalized checkpoints write their meta data out to persistent storage and are
- * <strong>not</strong> automatically cleaned up when the owning job fails or is suspended
- * (terminating with job status {@link JobStatus#FAILED} or {@link JobStatus#SUSPENDED}). In
- * this case, you have to manually clean up the checkpoint state, both the meta data and actual
- * program state.
- *
- * <p>The {@link ExternalizedCheckpointCleanup} mode defines how an externalized checkpoint
- * should be cleaned up on job cancellation. If you choose to retain externalized checkpoints on
- * cancellation you have to handle checkpoint clean-up manually when you cancel the job as well
- * (terminating with job status {@link JobStatus#CANCELED}).
- *
- * <p>The target directory for externalized checkpoints is configured via {@link
- * CheckpointingOptions#CHECKPOINTS_DIRECTORY}.
- *
- * @param cleanupMode Externalized checkpoint clean-up behaviour.
- * @deprecated Use {@link #setExternalizedCheckpointRetention} instead.
- */
- @Deprecated
- @PublicEvolving
- public void setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup cleanupMode) {
- configuration.set(ExecutionCheckpointingOptions.EXTERNALIZED_CHECKPOINT, cleanupMode);
- }
-
/**
* Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled
* automatically unless the mode is set to {@link
@@ -528,35 +497,6 @@ public class CheckpointConfig implements java.io.Serializable {
configuration.set(CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, cleanupMode);
}
- /**
- * Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled
- * automatically unless the mode is set to {@link
- * ExternalizedCheckpointCleanup#NO_EXTERNALIZED_CHECKPOINTS}.
- *
- * <p>Externalized checkpoints write their meta data out to persistent storage and are
- * <strong>not</strong> automatically cleaned up when the owning job fails or is suspended
- * (terminating with job status {@link JobStatus#FAILED} or {@link JobStatus#SUSPENDED}). In
- * this case, you have to manually clean up the checkpoint state, both the meta data and actual
- * program state.
- *
- * <p>The {@link ExternalizedCheckpointCleanup} mode defines how an externalized checkpoint
- * should be cleaned up on job cancellation. If you choose to retain externalized checkpoints on
- * cancellation you have to handle checkpoint clean-up manually when you cancel the job as well
- * (terminating with job status {@link JobStatus#CANCELED}).
- *
- * <p>The target directory for externalized checkpoints is configured via {@link
- * CheckpointingOptions#CHECKPOINTS_DIRECTORY}.
- *
- * @param cleanupMode Externalized checkpoint clean-up behaviour.
- * @deprecated use {@link #setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup)}
- * instead.
- */
- @PublicEvolving
- @Deprecated
- public void enableExternalizedCheckpoints(ExternalizedCheckpointCleanup cleanupMode) {
- setExternalizedCheckpointCleanup(cleanupMode);
- }
-
/**
* Returns whether checkpoints should be persisted externally.
*
@@ -564,8 +504,8 @@ public class CheckpointConfig implements java.io.Serializable {
*/
@PublicEvolving
public boolean isExternalizedCheckpointsEnabled() {
- return getExternalizedCheckpointCleanup()
- != ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS;
+ return getExternalizedCheckpointRetention()
+ != ExternalizedCheckpointRetention.NO_EXTERNALIZED_CHECKPOINTS;
}
/**
@@ -735,19 +675,6 @@ public class CheckpointConfig implements java.io.Serializable {
configuration.set(StateRecoveryOptions.APPROXIMATE_LOCAL_RECOVERY, enabled);
}
- /**
- * Returns the cleanup behaviour for externalized checkpoints.
- *
- * @return The cleanup behaviour for externalized checkpoints or <code>null</code> if none is
- * configured.
- * @deprecated Use {@link #getExternalizedCheckpointRetention} instead.
- */
- @Deprecated
- @PublicEvolving
- public ExternalizedCheckpointCleanup getExternalizedCheckpointCleanup() {
- return configuration.get(ExecutionCheckpointingOptions.EXTERNALIZED_CHECKPOINT);
- }
-
/**
* Returns the cleanup behaviour for externalized checkpoints.
*
@@ -783,68 +710,6 @@ public class CheckpointConfig implements java.io.Serializable {
return configuration.get(StateRecoveryOptions.CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA);
}
- /**
- * Cleanup behaviour for externalized checkpoints when the job is cancelled.
- *
- * @deprecated This class has been moved to {@link ExternalizedCheckpointRetention}.
- */
- @Deprecated
- @PublicEvolving
- public enum ExternalizedCheckpointCleanup implements DescribedEnum {
-
- /**
- * Delete externalized checkpoints on job cancellation.
- *
- * <p>All checkpoint state will be deleted when you cancel the owning job, both the meta
- * data and actual program state. Therefore, you cannot resume from externalized checkpoints
- * after the job has been cancelled.
- *
- * <p>Note that checkpoint state is always kept if the job terminates with state {@link
- * JobStatus#FAILED}.
- */
- DELETE_ON_CANCELLATION(
- text(
- "Checkpoint state is only kept when the owning job fails. It is deleted if "
- + "the job is cancelled.")),
-
- /**
- * Retain externalized checkpoints on job cancellation.
- *
- * <p>All checkpoint state is kept when you cancel the owning job. You have to manually
- * delete both the checkpoint meta data and actual program state after cancelling the job.
- *
- * <p>Note that checkpoint state is always kept if the job terminates with state {@link
- * JobStatus#FAILED}.
- */
- RETAIN_ON_CANCELLATION(
- text("Checkpoint state is kept when the owning job is cancelled or fails.")),
-
- /** Externalized checkpoints are disabled completely. */
- NO_EXTERNALIZED_CHECKPOINTS(text("Externalized checkpoints are disabled."));
-
- private final InlineElement description;
-
- ExternalizedCheckpointCleanup(InlineElement description) {
- this.description = description;
- }
-
- /**
- * Returns whether persistent checkpoints shall be discarded on cancellation of the job.
- *
- * @return <code>true</code> if persistent checkpoints shall be discarded on cancellation of
- * the job.
- */
- public boolean deleteOnCancellation() {
- return this == DELETE_ON_CANCELLATION;
- }
-
- @Override
- @Internal
- public InlineElement getDescription() {
- return description;
- }
- }
-
/**
* Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link
* CheckpointingOptions#CHECKPOINTING_CONSISTENCY_MODE}.
@@ -876,9 +741,6 @@ public class CheckpointConfig implements java.io.Serializable {
configuration
.getOptional(CheckpointingOptions.TOLERABLE_FAILURE_NUMBER)
.ifPresent(this::setTolerableCheckpointFailureNumber);
- configuration
- .getOptional(ExecutionCheckpointingOptions.EXTERNALIZED_CHECKPOINT)
- .ifPresent(this::setExternalizedCheckpointCleanup);
configuration
.getOptional(CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION)
.ifPresent(this::setExternalizedCheckpointRetention);
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
index de647c5568e..a15a7343802 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
@@ -112,41 +112,6 @@ public class ExecutionCheckpointingOptions {
+ "originating from the sync phase on the Task Managers are always forcing failover of an affected task. Other types of "
+ "checkpoint failures (such as checkpoint being subsumed) are being ignored.");
- @Deprecated
- @Documentation.ExcludeFromDocumentation("Hidden for deprecated.")
- public static final ConfigOption<CheckpointConfig.ExternalizedCheckpointCleanup>
- EXTERNALIZED_CHECKPOINT =
- ConfigOptions.key("execution.checkpointing.externalized-checkpoint-retention")
- .enumType(CheckpointConfig.ExternalizedCheckpointCleanup.class)
- .defaultValue(
- CheckpointConfig.ExternalizedCheckpointCleanup
- .NO_EXTERNALIZED_CHECKPOINTS)
- .withDescription(
- Description.builder()
- .text(
- "Externalized checkpoints write their meta data out to persistent storage and are not "
- + "automatically cleaned up when the owning job fails or is suspended (terminating with job "
- + "status %s or %s). In this case, you have to manually clean up the checkpoint state, both the "
- + "meta data and actual program state.",
- TextElement.code("JobStatus#FAILED"),
- TextElement.code("JobStatus#SUSPENDED"))
- .linebreak()
- .linebreak()
- .text(
- "The mode defines how an externalized checkpoint should be cleaned up on job cancellation. If "
- + "you choose to retain externalized checkpoints on cancellation you have to handle checkpoint "
- + "clean up manually when you cancel the job as well (terminating with job status %s).",
- TextElement.code("JobStatus#CANCELED"))
- .linebreak()
- .linebreak()
- .text(
- "The target directory for externalized checkpoints is configured via %s.",
- TextElement.code(
- CheckpointingOptions
- .CHECKPOINTS_DIRECTORY
- .key()))
- .build());
-
public static final ConfigOption<ExternalizedCheckpointRetention>
EXTERNALIZED_CHECKPOINT_RETENTION =
ConfigOptions.key("execution.checkpointing.externalized-checkpoint-retention")
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java
index fd73158fc68..a89a4ee4a29 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java
@@ -91,17 +91,6 @@ public class CheckpointConfigFromConfigurationTest {
.viaSetter(CheckpointConfig::setMinPauseBetweenCheckpoints)
.getterVia(CheckpointConfig::getMinPauseBetweenCheckpoints)
.nonDefaultValue(100L),
- TestSpec.testValue(
- CheckpointConfig.ExternalizedCheckpointCleanup
- .RETAIN_ON_CANCELLATION)
- .whenSetFromFile(
- "execution.checkpointing.externalized-checkpoint-retention",
- "RETAIN_ON_CANCELLATION")
- .viaSetter(CheckpointConfig::setExternalizedCheckpointCleanup)
- .getterVia(CheckpointConfig::getExternalizedCheckpointCleanup)
- .nonDefaultValue(
- CheckpointConfig.ExternalizedCheckpointCleanup
- .DELETE_ON_CANCELLATION),
TestSpec.testValue(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
.whenSetFromFile(
"execution.checkpointing.externalized-checkpoint-retention",
@@ -109,33 +98,6 @@ public class CheckpointConfigFromConfigurationTest {
.viaSetter(CheckpointConfig::setExternalizedCheckpointRetention)
.getterVia(CheckpointConfig::getExternalizedCheckpointRetention)
.nonDefaultValue(ExternalizedCheckpointRetention.DELETE_ON_CANCELLATION),
- TestSpec.testValue(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
- .whenSetFromFile(
- "execution.checkpointing.externalized-checkpoint-retention",
- "RETAIN_ON_CANCELLATION")
- .viaSetter(
- (config, v) -> {
- config.setExternalizedCheckpointCleanup(
- CheckpointConfig.ExternalizedCheckpointCleanup.valueOf(
- v.name()));
- })
- .getterVia(CheckpointConfig::getExternalizedCheckpointRetention)
- .nonDefaultValue(ExternalizedCheckpointRetention.DELETE_ON_CANCELLATION),
- TestSpec.testValue(
- CheckpointConfig.ExternalizedCheckpointCleanup
- .RETAIN_ON_CANCELLATION)
- .whenSetFromFile(
- "execution.checkpointing.externalized-checkpoint-retention",
- "RETAIN_ON_CANCELLATION")
- .viaSetter(
- (config, v) -> {
- config.setExternalizedCheckpointRetention(
- ExternalizedCheckpointRetention.valueOf(v.name()));
- })
- .getterVia(CheckpointConfig::getExternalizedCheckpointCleanup)
- .nonDefaultValue(
- CheckpointConfig.ExternalizedCheckpointCleanup
- .DELETE_ON_CANCELLATION),
TestSpec.testValue(12)
.whenSetFromFile(
"execution.checkpointing.tolerable-failed-checkpoints", "12")