class LocalFileCheckpointHandler implements CheckpointHandler {

  static LocalFileCheckpointHandler fromConfiguration() {
    checkState(Configuration.isInitialized(), "Configuration object not initialized");
    return new LocalFileCheckpointHandler(
        Configuration.getString(CONNECTOR_CHECKPOINT_DIRECTORY, DEFAULT_CHECKPOINT_DIRECTORY).get(),
        new FileHelper());
  }

  @VisibleForTesting  LocalFileCheckpointHandler(String directory, FileHelper fileHelper);

}