You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2022/08/05 06:39:55 UTC
[flink] 02/02: Revert "[FLINK-25745] Do not log exception if CompletedCheckpointStorageLocation#disposeStorageLocation failed in CLAIM mode"
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 17a782c202c93343b8884cb52f4562f9c4ba593f
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Fri Jul 22 14:08:29 2022 +0200
Revert "[FLINK-25745] Do not log exception if CompletedCheckpointStorageLocation#disposeStorageLocation failed in CLAIM mode"
This reverts commit d91cb003
---
.../runtime/checkpoint/CheckpointCoordinator.java | 3 +-
.../flink/runtime/checkpoint/Checkpoints.java | 41 ++--------------------
.../checkpoint/CheckpointMetadataLoadingTest.java | 16 +++------
3 files changed, 8 insertions(+), 52 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 386bb07ee97..618f7d899ca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -1799,8 +1799,7 @@ public class CheckpointCoordinator {
checkpointLocation,
userClassLoader,
allowNonRestored,
- checkpointProperties,
- restoreSettings.getRestoreMode());
+ checkpointProperties);
// register shared state - even before adding the checkpoint to the store
// because the latter might trigger subsumption so the ref counts must be up-to-date
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java
index 5443301363b..736393f8dd9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.checkpoint.metadata.MetadataV4Serializer;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStorageLoader;
import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
@@ -132,8 +131,7 @@ public class Checkpoints {
CompletedCheckpointStorageLocation location,
ClassLoader classLoader,
boolean allowNonRestoredState,
- CheckpointProperties checkpointProperties,
- RestoreMode restoreMode)
+ CheckpointProperties checkpointProperties)
throws IOException {
checkNotNull(jobId, "jobId");
@@ -221,9 +219,7 @@ public class Checkpoints {
operatorStates,
checkpointMetadata.getMasterStates(),
checkpointProperties,
- restoreMode == RestoreMode.CLAIM
- ? new ClaimModeCompletedStorageLocation(location)
- : location,
+ location,
null,
checkpointMetadata.getCheckpointProperties());
}
@@ -383,37 +379,4 @@ public class Checkpoints {
/** This class contains only static utility methods and is not meant to be instantiated. */
private Checkpoints() {}
-
- private static class ClaimModeCompletedStorageLocation
- implements CompletedCheckpointStorageLocation {
-
- private final CompletedCheckpointStorageLocation wrapped;
-
- private ClaimModeCompletedStorageLocation(CompletedCheckpointStorageLocation location) {
- wrapped = location;
- }
-
- @Override
- public String getExternalPointer() {
- return wrapped.getExternalPointer();
- }
-
- @Override
- public StreamStateHandle getMetadataHandle() {
- return wrapped.getMetadataHandle();
- }
-
- @Override
- public void disposeStorageLocation() throws IOException {
- try {
- wrapped.disposeStorageLocation();
- } catch (Exception ex) {
- LOG.debug(
- "We could not delete the storage location: {} in CLAIM restore mode. It is"
- + " most probably because of shared files still being used by newer"
- + " checkpoints",
- wrapped);
- }
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
index 9583ccbfef3..2632ef703b4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java
@@ -25,7 +25,6 @@ import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.OperatorStreamStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
@@ -79,8 +78,7 @@ public class CheckpointMetadataLoadingTest {
testSavepoint,
cl,
false,
- CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL),
- RestoreMode.NO_CLAIM);
+ CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL));
assertEquals(jobId, loaded.getJobId());
assertEquals(checkpointId, loaded.getCheckpointID());
@@ -104,8 +102,7 @@ public class CheckpointMetadataLoadingTest {
testSavepoint,
cl,
false,
- CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL),
- RestoreMode.NO_CLAIM);
+ CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL));
fail("Did not throw expected Exception");
} catch (IllegalStateException expected) {
assertTrue(expected.getMessage().contains("Max parallelism mismatch"));
@@ -131,8 +128,7 @@ public class CheckpointMetadataLoadingTest {
testSavepoint,
cl,
false,
- CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL),
- RestoreMode.NO_CLAIM);
+ CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL));
fail("Did not throw expected Exception");
} catch (IllegalStateException expected) {
assertTrue(expected.getMessage().contains("allowNonRestoredState"));
@@ -158,8 +154,7 @@ public class CheckpointMetadataLoadingTest {
testSavepoint,
cl,
true,
- CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL),
- RestoreMode.NO_CLAIM);
+ CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL));
assertTrue(loaded.getOperatorStates().isEmpty());
}
@@ -188,8 +183,7 @@ public class CheckpointMetadataLoadingTest {
testSavepoint,
cl,
false,
- CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL),
- RestoreMode.NO_CLAIM);
+ CheckpointProperties.forSavepoint(false, SavepointFormatType.CANONICAL));
fail("Did not throw expected Exception");
} catch (IllegalStateException expected) {
assertTrue(expected.getMessage().contains("allowNonRestoredState"));