[FLINK-9751] Add a RecoverableWriter to the FileSystem abstraction

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.6.0
• Component/s: Connectors / Common
• Assignee: Stephan Ewen
• Reporter: Stephan Ewen
• Created: 04/Jul/18 20:13
• Updated: 15/Jul/18 22:24
• Resolved: 15/Jul/18 22:24

Description

The core operation of the StreamingFileSink is to append result data to (hidden) "in progress" files and then, when the files should roll over, publish them as visible files. At each checkpoint, the data so far must be persistent in the "in progress" files. On recovery, we resume the "in progress" file at the exact position of the checkpoint, or publish up to the position of that checkpoint.

In order to support various file systems and object stores, we need an interface that captures these core operations and allows for different implementations (such as file truncate/append on posix, MultiPartUpload on S3, ...)

Proposed interface:

/**
 * A handle to an in-progress stream with a defined and persistent amount of data.
 * The handle can be used to recover the stream and publish the result file.
 */
interface CommitRecoverable { ... }

/**
 * A handle to an in-progress stream with a defined and persistent amount of data.
 * The handle can be used to recover the stream and either publish the result file
 * or keep appending data to the stream.
 */
interface ResumeRecoverable extends CommitRecoverable { ... }

/**
 * An output stream to a file system that can be recovered at well defined points.
 * The stream initially writes to hidden files or temp files and only creates the
 * target file once it is closed and "committed".
 */
public abstract class RecoverableFsDataOutputStream extends FSDataOutputStream {

    /**
     * Ensures all data so far is persistent (similar to {@link #sync()}) and returns
     * a handle to recover the stream at the current position.
     */
    public abstract ResumeRecoverable persist() throws IOException;

    /**
     * Closes the stream, ensuring persistence of all data (similar to {@link #sync()}).
     * This returns a Committer that can be used to publish (make visible) the file
     * that the stream was writing to.
     */
    public abstract Committer closeForCommit() throws IOException;

    /**
     * A committer can publish the file of a stream that was closed.
     * The Committer can be recovered via a {@link CommitRecoverable}.
     */
    public interface Committer {

        void commit() throws IOException;

        CommitRecoverable getRecoverable();
    }
}

/**
 * The RecoverableWriter creates and recovers RecoverableFsDataOutputStream.
 */
public interface RecoverableWriter{

    RecoverableFsDataOutputStream open(Path path) throws IOException;

    RecoverableFsDataOutputStream recover(ResumeRecoverable resumable) throws IOException;

    RecoverableFsDataOutputStream.Committer recoverForCommit(CommitRecoverable resumable) throws IOException;