@@ -0,0 +1,2585 @@
+package tachyon.worker.block.io;%0A%0Aimport java.io.File;%0Aimport java.io.FileNotFoundException;%0Aimport java.io.IOException;%0Aimport java.io.RandomAccessFile;%0Aimport java.nio.ByteBuffer;%0Aimport java.nio.channels.ByteChannel;%0Aimport java.nio.channels.FileChannel;%0Aimport java.nio.channels.WritableByteChannel;%0A%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport com.google.common.base.Preconditions;%0Aimport com.google.common.io.Closer;%0A%0Aimport tachyon.Constants;%0Aimport tachyon.util.CommonUtils;%0Aimport tachyon.worker.block.meta.BlockMeta;%0A%0A/**%0A * This class provides write access to a block data file locally stored in managed storage.%0A * %3Cp%3E%0A * This class does not provide thread-safety. Corresponding lock must be acquired.%0A */%0Apublic class LocalFileBlockWriter implements BlockWriter %7B%0A  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);%0A  private final BlockMeta mBlockMeta;%0A  private final String mFilePath;%0A  private final RandomAccessFile mLocalFile;%0A  private final FileChannel mLocalFileChannel;%0A  private final Closer mCloser = Closer.create();%0A%0A  public LocalFileBlockWriter(BlockMeta blockMeta) throws IOException %7B%0A    mBlockMeta = Preconditions.checkNotNull(blockMeta);%0A    mFilePath = mBlockMeta.getPath();%0A    mLocalFile = mCloser.register(new RandomAccessFile(mFilePath, %22w%22));%0A    mLocalFileChannel = mCloser.register(mLocalFile.getChannel());%0A  %7D%0A%0A  @Override%0A  public WritableByteChannel getChannel() %7B%0A    return mLocalFileChannel;%0A  %7D%0A%0A  @Override%0A  public long append(ByteBuffer inputBuf) throws IOException %7B%0A    return write(mLocalFileChannel.size(), inputBuf);%0A  %7D%0A%0A  /**%0A   * Writes data to the block from an input ByteBuffer.%0A   *%0A   * @param offset starting offset of the block file to write%0A   * @param inputBuf ByteBuffer that input data is stored in%0A   * @return the size of data that was written%0A   * @throws IOException%0A   */%0A  private long write(long offset, ByteBuffer inputBuf) throws IOException %7B%0A    int inputBufLength = inputBuf.limit();%0A    ByteBuffer outputBuf =%0A        mLocalFileChannel.map(FileChannel.MapMode.READ_WRITE, offset, inputBufLength);%0A    outputBuf.put(inputBuf);%0A    CommonUtils.cleanDirectBuffer(outputBuf);%0A    return outputBuf.limit();%0A  %7D%0A%0A  @Override%0A  public boolean delete() %7B%0A    return new File(mFilePath).delete();%0A  %7D%0A%0A  @Override%0A  public boolean move(String dstPath) %7B%0A    // Check if destPath equals the current path, in this case, do nothing.%0A    if (mFilePath.equals(dstPath)) %7B%0A      return true;%0A    %7D%0A    return new File(mFilePath).renameTo(new File(dstPath));%0A  %7D%0A%7D

Add a implementation of BlockWriter based on local file