[ https://issues.apache.org/jira/browse/FLINK-4512?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15565540#comment-15565540 ]
ASF GitHub Bot commented on FLINK-4512: --------------------------------------- Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/2608#discussion_r82796670 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java --- @@ -18,23 +18,105 @@ package org.apache.flink.runtime.checkpoint.savepoint; +import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.core.fs.FSDataOutputStream; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.Path; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.util.FileUtils; +import org.apache.flink.util.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +import static org.apache.flink.util.Preconditions.checkNotNull; + /** - * Savepoint store used to persist {@link Savepoint} instances. + * A file system based savepoint store. * - * <p>The main implementation is the {@link FsSavepointStore}. We also have the - * {@link HeapSavepointStore} for historical reasons (introduced in Flink 1.0). + * <p>Stored savepoints have the following format: + * <pre> + * MagicNumber SavepointVersion Savepoint + * - MagicNumber => int + * - SavepointVersion => int (returned by Savepoint#getVersion()) + * - Savepoint => bytes (serialized via version-specific SavepointSerializer) + * </pre> */ -public interface SavepointStore { +public class SavepointStore { + + private static final Logger LOG = LoggerFactory.getLogger(SavepointStore.class); + + /** Magic number for sanity checks against stored savepoints. */ + private static final int MAGIC_NUMBER = 0x4960672d; + + /** Prefix for savepoint files. */ + private static final String prefix = "savepoint-"; /** * Stores the savepoint. * + * @param targetDirectory Target directory to store savepoint in * @param savepoint Savepoint to be stored * @param <T> Savepoint type * @return Path of stored savepoint * @throws Exception Failures during store are forwarded */ - <T extends Savepoint> String storeSavepoint(T savepoint) throws Exception; + public static <T extends Savepoint> String storeSavepoint( + String targetDirectory, + T savepoint) throws IOException { + + checkNotNull(targetDirectory, "Target directory"); + checkNotNull(savepoint, "Savepoint"); + + Exception latestException = null; + Path path = null; + FSDataOutputStream fdos = null; + + FileSystem fs = null; + + // Try to create a FS output stream + for (int attempt = 0; attempt < 10; attempt++) { + path = new Path(targetDirectory, FileUtils.getRandomFilename(prefix)); + + if (fs == null) { + fs = FileSystem.get(path.toUri()); + } + + try { + fdos = fs.create(path, false); + break; + } catch (Exception e) { + latestException = e; + } + } + + if (fdos == null) { + throw new IOException("Failed to create file output stream at " + path, latestException); + } + + boolean success = false; + try (DataOutputStream dos = new DataOutputStream(fdos)) { + // Write header + dos.writeInt(MAGIC_NUMBER); + dos.writeInt(savepoint.getVersion()); + + // Write savepoint + SavepointSerializer<T> serializer = SavepointSerializers.getSerializer(savepoint); + serializer.serialize(savepoint, dos); + success = true; + } finally { + if (!success && fs.exists(path)) { + if (!fs.delete(path, true)) { + LOG.warn("Failed to delete file " + path + " after failed write."); --- End diff -- Placeholder {} > Add option for persistent checkpoints > ------------------------------------- > > Key: FLINK-4512 > URL: https://issues.apache.org/jira/browse/FLINK-4512 > Project: Flink > Issue Type: Sub-task > Components: State Backends, Checkpointing > Reporter: Ufuk Celebi > Assignee: Ufuk Celebi > > Allow periodic checkpoints to be persisted by writing out their meta data. > This is what we currently do for savepoints, but in the future checkpoints > and savepoints are likely to diverge with respect to guarantees they give for > updatability, etc. > This means that the difference between persistent checkpoints and savepoints > in the long term will be that persistent checkpoints can only be restored > with the same job settings (like parallelism, etc.) > Regular and persisted checkpoints should behave differently with respect to > disposal in *globally* terminal job states (FINISHED, CANCELLED, FAILED): > regular checkpoints are cleaned up in all of these cases whereas persistent > checkpoints only on FINISHED. Maybe with the option to customize behaviour on > CANCELLED or FAILED. -- This message was sent by Atlassian JIRA (v6.3.4#6332)