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 <[email protected]> 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"));
