content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.apex.malhar.flume.storage;
  
  import java.io.DataInputStream;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  import javax.validation.constraints.NotNull;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.flume.sink.Server;
  
  import org.apache.flume.Context;
  import org.apache.flume.conf.Configurable;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  
  import com.google.common.primitives.Ints;
  import com.google.common.primitives.Longs;
  
  import com.datatorrent.api.Component;
  import com.datatorrent.common.util.NameableThreadFactory;
  import com.datatorrent.netlet.util.Slice;
  
  /**
   * HDFSStorage is developed to store and retrieve the data from HDFS
   * <p />
   * The properties that can be set on HDFSStorage are: <br />
   * baseDir - The base directory where the data is going to be stored <br />
   * restore - This is used to restore the application from previous failure <br />
   * blockSize - The maximum size of the each file to created. <br />
   *
   * @since 0.9.3
   */
  public class HDFSStorage implements Storage, Configurable, Component<com.datatorrent.api.Context>
  {
    public static final int DEFAULT_BLOCK_SIZE = 64 * 1024 * 1024;
    public static final String BASE_DIR_KEY = "baseDir";
    public static final String RESTORE_KEY = "restore";
    public static final String BLOCKSIZE = "blockSize";
    public static final String BLOCK_SIZE_MULTIPLE = "blockSizeMultiple";
    public static final String NUMBER_RETRY = "retryCount";
  
    private static final String OFFSET_SUFFIX = "-offsetFile";
    private static final String BOOK_KEEPING_FILE_OFFSET = "-bookKeepingOffsetFile";
    private static final String FLUSHED_IDENTITY_FILE = "flushedCounter";
    private static final String CLEAN_OFFSET_FILE = "cleanoffsetFile";
    private static final String FLUSHED_IDENTITY_FILE_TEMP = "flushedCounter.tmp";
    private static final String CLEAN_OFFSET_FILE_TEMP = "cleanoffsetFile.tmp";
    private static final int IDENTIFIER_SIZE = 8;
    private static final int DATA_LENGTH_BYTE_SIZE = 4;
  
    /**
     * Number of times the storage will try to get the filesystem
     */
    private int retryCount = 3;
    /**
     * The multiple of block size
     */
    private int blockSizeMultiple = 1;
    /**
     * Identifier for this storage.
     */
    @NotNull
    private String id;
    /**
     * The baseDir where the storage facility is going to create files.
     */
    @NotNull
    private String baseDir;
    /**
     * The block size to be used to create the storage files
     */
    private long blockSize;
    /**
     *
     */
    private boolean restore;
    /**
     * This identifies the current file number
     */
    private long currentWrittenFile;
    /**
     * This identifies the file number that has been flushed
     */
    private long flushedFileCounter;
    /**
     * The file that stores the fileCounter information
     */
    // private Path fileCounterFile;
    /**
     * The file that stores the flushed fileCounter information
     */
    private Path flushedCounterFile;
    private Path flushedCounterFileTemp;
    /**
     * This identifies the last cleaned file number
     */
    private long cleanedFileCounter;
    /**
     * The file that stores the clean file counter information
     */
    // private Path cleanFileCounterFile;
    /**
     * The file that stores the clean file offset information
     */
    private Path cleanFileOffsetFile;
    private Path cleanFileOffsetFileTemp;
    private FileSystem fs;
    private FSDataOutputStream dataStream;
    ArrayList<DataBlock> files2Commit = new ArrayList<DataBlock>();
    /**
     * The offset in the current opened file
     */
    private long fileWriteOffset;
    private FSDataInputStream readStream;
    private long retrievalOffset;
    private long retrievalFile;
    private int offset;
    private long flushedLong;
    private long flushedFileWriteOffset;
    private long bookKeepingFileOffset;
    private byte[] cleanedOffset = new byte[8];
    private long skipOffset;
    private long skipFile;
    private transient Path basePath;
    private ExecutorService storageExecutor;
    private byte[] currentData;
    private FSDataInputStream nextReadStream;
    private long nextFlushedLong;
    private long nextRetrievalFile;
    private byte[] nextRetrievalData;
  
    public HDFSStorage()
    {
      this.restore = true;
    }
  
    /**
     * This stores the Identifier information identified in the last store function call
     *
     * @param ctx
     */
    @Override
    public void configure(Context ctx)
    {
      String tempId = ctx.getString(ID);
      if (tempId == null) {
        if (id == null) {
          throw new IllegalArgumentException("id can't be  null.");
        }
      } else {
        id = tempId;
      }
  
      String tempBaseDir = ctx.getString(BASE_DIR_KEY);
      if (tempBaseDir != null) {
        baseDir = tempBaseDir;
      }
  
      restore = ctx.getBoolean(RESTORE_KEY, restore);
      Long tempBlockSize = ctx.getLong(BLOCKSIZE);
      if (tempBlockSize != null) {
        blockSize = tempBlockSize;
      }
      blockSizeMultiple = ctx.getInteger(BLOCK_SIZE_MULTIPLE, blockSizeMultiple);
      retryCount = ctx.getInteger(NUMBER_RETRY,retryCount);
    }
  
    /**
     * This function reads the file at a location and return the bytes stored in the file "
     *
     * @param path - the location of the file
     * @return
     * @throws IOException
     */
    byte[] readData(Path path) throws IOException
    {
      DataInputStream is = new DataInputStream(fs.open(path));
      byte[] bytes = new byte[is.available()];
      is.readFully(bytes);
      is.close();
      return bytes;
    }
  
    /**
     * This function writes the bytes to a file specified by the path
     *
     * @param path the file location
     * @param data the data to be written to the file
     * @return
     * @throws IOException
     */
    private FSDataOutputStream writeData(Path path, byte[] data) throws IOException
    {
      FSDataOutputStream fsOutputStream;
      if (fs.getScheme().equals("file")) {
        // local FS does not support hflush and does not flush native stream
        fsOutputStream = new FSDataOutputStream(
            new FileOutputStream(Path.getPathWithoutSchemeAndAuthority(path).toString()), null);
      } else {
        fsOutputStream = fs.create(path);
      }
      fsOutputStream.write(data);
      return fsOutputStream;
    }
  
    private long calculateOffset(long fileOffset, long fileCounter)
    {
      return ((fileCounter << 32) | (fileOffset & 0xffffffffL));
    }
  
    @Override
    public byte[] store(Slice slice)
    {
      // logger.debug("store message ");
      int bytesToWrite = slice.length + DATA_LENGTH_BYTE_SIZE;
      if (currentWrittenFile < skipFile) {
        fileWriteOffset += bytesToWrite;
        if (fileWriteOffset >= bookKeepingFileOffset) {
          files2Commit.add(new DataBlock(null, bookKeepingFileOffset,
              new Path(basePath, currentWrittenFile + OFFSET_SUFFIX), currentWrittenFile));
          currentWrittenFile++;
          if (fileWriteOffset > bookKeepingFileOffset) {
            fileWriteOffset = bytesToWrite;
          } else {
            fileWriteOffset = 0;
          }
          try {
            bookKeepingFileOffset = getFlushedFileWriteOffset(
                new Path(basePath, currentWrittenFile + BOOK_KEEPING_FILE_OFFSET));
          } catch (IOException e) {
            throw new RuntimeException(e);
          }
        }
        return null;
      }
  
      if (flushedFileCounter == currentWrittenFile && dataStream == null) {
        currentWrittenFile++;
        fileWriteOffset = 0;
      }
  
      if (flushedFileCounter == skipFile && skipFile != -1) {
        skipFile++;
      }
  
      if (fileWriteOffset + bytesToWrite < blockSize) {
        try {
          /* write length and the actual data to the file */
          if (fileWriteOffset == 0) {
            // writeData(flushedCounterFile, String.valueOf(currentWrittenFile).getBytes()).close();
            dataStream = writeData(new Path(basePath, String.valueOf(currentWrittenFile)),
                Ints.toByteArray(slice.length));
            dataStream.write(slice.buffer, slice.offset, slice.length);
          } else {
            dataStream.write(Ints.toByteArray(slice.length));
            dataStream.write(slice.buffer, slice.offset, slice.length);
          }
          fileWriteOffset += bytesToWrite;
  
          byte[] fileOffset = null;
          if ((currentWrittenFile > skipFile) || (currentWrittenFile == skipFile && fileWriteOffset > skipOffset)) {
            skipFile = -1;
            fileOffset = new byte[IDENTIFIER_SIZE];
            Server.writeLong(fileOffset, 0, calculateOffset(fileWriteOffset, currentWrittenFile));
          }
          return fileOffset;
        } catch (IOException ex) {
          logger.warn("Error while storing the bytes {}", ex.getMessage());
          closeFs();
          throw new RuntimeException(ex);
        }
      }
      DataBlock db = new DataBlock(dataStream, fileWriteOffset,
          new Path(basePath, currentWrittenFile + OFFSET_SUFFIX), currentWrittenFile);
      db.close();
      files2Commit.add(db);
      fileWriteOffset = 0;
      ++currentWrittenFile;
      return store(slice);
    }
  
    /**
     * @param b
     * @param startIndex
     * @return
     */
    long byteArrayToLong(byte[] b, int startIndex)
    {
      final byte b1 = 0;
      return Longs.fromBytes(b1, b1, b1, b1, b[3 + startIndex], b[2 + startIndex], b[1 + startIndex], b[startIndex]);
    }
  
    @Override
    public byte[] retrieve(byte[] identifier)
    {
      skipFile = -1;
      skipOffset = 0;
      logger.debug("retrieve with address {}", Arrays.toString(identifier));
      // flushing the last incomplete flushed file
      closeUnflushedFiles();
  
      retrievalOffset = byteArrayToLong(identifier, 0);
      retrievalFile = byteArrayToLong(identifier, offset);
  
      if (retrievalFile == 0 && retrievalOffset == 0 && currentWrittenFile == 0 && fileWriteOffset == 0) {
        skipOffset = 0;
        return null;
      }
  
      // making sure that the deleted address is not requested again
      if (retrievalFile != 0 || retrievalOffset != 0) {
        long cleanedFile = byteArrayToLong(cleanedOffset, offset);
        if (retrievalFile < cleanedFile || (retrievalFile == cleanedFile &&
            retrievalOffset < byteArrayToLong(cleanedOffset, 0))) {
          logger.warn("The address asked has been deleted retrievalFile={}, cleanedFile={}, retrievalOffset={}, " +
              "cleanedOffset={}", retrievalFile, cleanedFile, retrievalOffset, byteArrayToLong(cleanedOffset, 0));
          closeFs();
          throw new IllegalArgumentException(String.format("The data for address %s has already been deleted",
              Arrays.toString(identifier)));
        }
      }
  
      // we have just started
      if (retrievalFile == 0 && retrievalOffset == 0) {
        retrievalFile = byteArrayToLong(cleanedOffset, offset);
        retrievalOffset = byteArrayToLong(cleanedOffset, 0);
      }
  
      if ((retrievalFile > flushedFileCounter)) {
        skipFile = retrievalFile;
        skipOffset = retrievalOffset;
        retrievalFile = -1;
        return null;
      }
      if ((retrievalFile == flushedFileCounter && retrievalOffset >= flushedFileWriteOffset)) {
        skipFile = retrievalFile;
        skipOffset = retrievalOffset - flushedFileWriteOffset;
        retrievalFile = -1;
        return null;
      }
  
      try {
        if (readStream != null) {
          readStream.close();
          readStream = null;
        }
        Path path = new Path(basePath, String.valueOf(retrievalFile));
        if (!fs.exists(path)) {
          retrievalFile = -1;
          closeFs();
          throw new RuntimeException(String.format("File %s does not exist", path.toString()));
        }
  
        byte[] flushedOffset = readData(new Path(basePath, retrievalFile + OFFSET_SUFFIX));
        flushedLong = Server.readLong(flushedOffset, 0);
        while (retrievalOffset >= flushedLong && retrievalFile < flushedFileCounter) {
          retrievalOffset -= flushedLong;
          retrievalFile++;
          flushedOffset = readData(new Path(basePath, retrievalFile + OFFSET_SUFFIX));
          flushedLong = Server.readLong(flushedOffset, 0);
        }
  
        if (retrievalOffset >= flushedLong) {
          logger.warn("data not flushed for the given identifier");
          retrievalFile = -1;
          return null;
        }
        synchronized (HDFSStorage.this) {
          if (nextReadStream != null) {
            nextReadStream.close();
            nextReadStream = null;
          }
        }
        currentData = null;
        path = new Path(basePath, String.valueOf(retrievalFile));
        //readStream = new FSDataInputStream(fs.open(path));
        currentData = readData(path);
        //readStream.seek(retrievalOffset);
        storageExecutor.submit(getNextStream());
        return retrieveHelper();
      } catch (IOException e) {
        closeFs();
        throw new RuntimeException(e);
      }
    }
  
    private byte[] retrieveHelper() throws IOException
    {
      int tempRetrievalOffset = (int)retrievalOffset;
      int length = Ints.fromBytes(currentData[tempRetrievalOffset], currentData[tempRetrievalOffset + 1],
          currentData[tempRetrievalOffset + 2], currentData[tempRetrievalOffset + 3]);
      byte[] data = new byte[length + IDENTIFIER_SIZE];
      System.arraycopy(currentData, tempRetrievalOffset + 4, data, IDENTIFIER_SIZE, length);
      retrievalOffset += length + DATA_LENGTH_BYTE_SIZE;
      if (retrievalOffset >= flushedLong) {
        Server.writeLong(data, 0, calculateOffset(0, retrievalFile + 1));
      } else {
        Server.writeLong(data, 0, calculateOffset(retrievalOffset, retrievalFile));
      }
      return data;
    }
  
    @Override
    public byte[] retrieveNext()
    {
      if (retrievalFile == -1) {
        closeFs();
        throw new RuntimeException("Call retrieve first");
      }
  
      if (retrievalFile > flushedFileCounter) {
        logger.warn("data is not flushed");
        return null;
      }
  
      try {
        if (currentData == null) {
          synchronized (HDFSStorage.this) {
            if (nextRetrievalData != null && (retrievalFile == nextRetrievalFile)) {
              currentData = nextRetrievalData;
              flushedLong = nextFlushedLong;
              nextRetrievalData = null;
            } else {
              currentData = null;
              currentData = readData(new Path(basePath, String.valueOf(retrievalFile)));
              byte[] flushedOffset = readData(new Path(basePath, retrievalFile + OFFSET_SUFFIX));
              flushedLong = Server.readLong(flushedOffset, 0);
            }
          }
          storageExecutor.submit(getNextStream());
        }
  
        if (retrievalOffset >= flushedLong) {
          retrievalFile++;
          retrievalOffset = 0;
  
          if (retrievalFile > flushedFileCounter) {
            logger.warn("data is not flushed");
            return null;
          }
  
          //readStream.close();
          // readStream = new FSDataInputStream(fs.open(new Path(basePath, String.valueOf(retrievalFile))));
          // byte[] flushedOffset = readData(new Path(basePath, retrievalFile + OFFSET_SUFFIX));
          // flushedLong = Server.readLong(flushedOffset, 0);
  
          synchronized (HDFSStorage.this) {
            if (nextRetrievalData != null && (retrievalFile == nextRetrievalFile)) {
              currentData = nextRetrievalData;
              flushedLong = nextFlushedLong;
              nextRetrievalData = null;
            } else {
              currentData = null;
              currentData = readData(new Path(basePath, String.valueOf(retrievalFile)));
              byte[] flushedOffset = readData(new Path(basePath, retrievalFile + OFFSET_SUFFIX));
              flushedLong = Server.readLong(flushedOffset, 0);
            }
          }
          storageExecutor.submit(getNextStream());
        }
        //readStream.seek(retrievalOffset);
        return retrieveHelper();
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  
    @Override
    @SuppressWarnings("AssignmentToCollectionOrArrayFieldFromParameter")
    public void clean(byte[] identifier)
    {
      logger.info("clean {}", Arrays.toString(identifier));
      long cleanFileIndex = byteArrayToLong(identifier, offset);
  
      long cleanFileOffset = byteArrayToLong(identifier, 0);
      if (flushedFileCounter == -1) {
        identifier = new byte[8];
      } else if (cleanFileIndex > flushedFileCounter ||
          (cleanFileIndex == flushedFileCounter && cleanFileOffset >= flushedFileWriteOffset)) {
        // This is to make sure that we clean only the data that is flushed
        cleanFileIndex = flushedFileCounter;
        cleanFileOffset = flushedFileWriteOffset;
        Server.writeLong(identifier, 0, calculateOffset(cleanFileOffset, cleanFileIndex));
      }
      cleanedOffset = identifier;
  
      try {
        writeData(cleanFileOffsetFileTemp, identifier).close();
        fs.rename(cleanFileOffsetFileTemp, cleanFileOffsetFile);
        if (cleanedFileCounter >= cleanFileIndex) {
          return;
        }
        do {
          Path path = new Path(basePath, String.valueOf(cleanedFileCounter));
          if (fs.exists(path) && fs.isFile(path)) {
            fs.delete(path, false);
          }
          path = new Path(basePath, cleanedFileCounter + OFFSET_SUFFIX);
          if (fs.exists(path) && fs.isFile(path)) {
            fs.delete(path, false);
          }
          path = new Path(basePath, cleanedFileCounter + BOOK_KEEPING_FILE_OFFSET);
          if (fs.exists(path) && fs.isFile(path)) {
            fs.delete(path, false);
          }
          logger.info("deleted file {}", cleanedFileCounter);
          ++cleanedFileCounter;
        }
        while (cleanedFileCounter < cleanFileIndex);
        // writeData(cleanFileCounterFile, String.valueOf(cleanedFileCounter).getBytes()).close();
  
      } catch (IOException e) {
        logger.warn("not able to close the streams {}", e.getMessage());
        closeFs();
        throw new RuntimeException(e);
      }
    }
  
    /**
     * This is used mainly for cleaning up of counter files created
     */
    void cleanHelperFiles()
    {
      try {
        fs.delete(basePath, true);
      } catch (IOException e) {
        logger.warn(e.getMessage());
      }
    }
  
    private void closeUnflushedFiles()
    {
      try {
        files2Commit.clear();
        // closing the stream
        if (dataStream != null) {
          dataStream.close();
          dataStream = null;
          // currentWrittenFile++;
          // fileWriteOffset = 0;
        }
  
        if (!fs.exists(new Path(basePath, currentWrittenFile + OFFSET_SUFFIX))) {
          fs.delete(new Path(basePath, String.valueOf(currentWrittenFile)), false);
        }
  
        if (fs.exists(new Path(basePath, flushedFileCounter + OFFSET_SUFFIX))) {
          // This means that flush was called
          flushedFileWriteOffset = getFlushedFileWriteOffset(new Path(basePath, flushedFileCounter + OFFSET_SUFFIX));
          bookKeepingFileOffset = getFlushedFileWriteOffset(
              new Path(basePath, flushedFileCounter + BOOK_KEEPING_FILE_OFFSET));
        }
  
        if (flushedFileCounter != -1) {
          currentWrittenFile = flushedFileCounter;
          fileWriteOffset = flushedFileWriteOffset;
        } else {
          currentWrittenFile = 0;
          fileWriteOffset = 0;
        }
  
        flushedLong = 0;
  
      } catch (IOException e) {
        closeFs();
        throw new RuntimeException(e);
      }
    }
  
    @Override
    public void flush()
    {
      nextReadStream = null;
      StringBuilder builder = new StringBuilder();
      Iterator<DataBlock> itr = files2Commit.iterator();
      DataBlock db;
      try {
        while (itr.hasNext()) {
          db = itr.next();
          db.updateOffsets();
          builder.append(db.fileName).append(", ");
        }
        files2Commit.clear();
  
        if (dataStream != null) {
          dataStream.hflush();
          writeData(flushedCounterFileTemp, String.valueOf(currentWrittenFile).getBytes()).close();
          fs.rename(flushedCounterFileTemp, flushedCounterFile);
          updateFlushedOffset(new Path(basePath, currentWrittenFile + OFFSET_SUFFIX), fileWriteOffset);
          flushedFileWriteOffset = fileWriteOffset;
          builder.append(currentWrittenFile);
        }
        logger.debug("flushed files {}", builder.toString());
      } catch (IOException ex) {
        logger.warn("not able to close the stream {}", ex.getMessage());
        closeFs();
        throw new RuntimeException(ex);
      }
      flushedFileCounter = currentWrittenFile;
      // logger.debug("flushedFileCounter in flush {}",flushedFileCounter);
    }
  
    /**
     * This updates the flushed offset
     */
    private void updateFlushedOffset(Path file, long bytesWritten)
    {
      byte[] lastStoredOffset = new byte[IDENTIFIER_SIZE];
      Server.writeLong(lastStoredOffset, 0, bytesWritten);
      try {
        writeData(file, lastStoredOffset).close();
      } catch (IOException e) {
        try {
          if (!Arrays.equals(readData(file), lastStoredOffset)) {
            closeFs();
            throw new RuntimeException(e);
          }
        } catch (Exception e1) {
          closeFs();
          throw new RuntimeException(e1);
        }
      }
    }
  
    public int getBlockSizeMultiple()
    {
      return blockSizeMultiple;
    }
  
    public void setBlockSizeMultiple(int blockSizeMultiple)
    {
      this.blockSizeMultiple = blockSizeMultiple;
    }
  
    /**
     * @return the baseDir
     */
    public String getBaseDir()
    {
      return baseDir;
    }
  
    /**
     * @param baseDir the baseDir to set
     */
    public void setBaseDir(String baseDir)
    {
      this.baseDir = baseDir;
    }
  
    /**
     * @return the id
     */
    public String getId()
    {
      return id;
    }
  
    /**
     * @param id the id to set
     */
    public void setId(String id)
    {
      this.id = id;
    }
  
    /**
     * @return the blockSize
     */
    public long getBlockSize()
    {
      return blockSize;
    }
  
    /**
     * @param blockSize the blockSize to set
     */
    public void setBlockSize(long blockSize)
    {
      this.blockSize = blockSize;
    }
  
    /**
     * @return the restore
     */
    public boolean isRestore()
    {
      return restore;
    }
  
    /**
     * @param restore the restore to set
     */
    public void setRestore(boolean restore)
    {
      this.restore = restore;
    }
  
    class DataBlock
    {
      FSDataOutputStream dataStream;
      long dataOffset;
      Path path2FlushedData;
      long fileName;
      private Path bookKeepingPath;
  
      DataBlock(FSDataOutputStream stream, long bytesWritten, Path path2FlushedData, long fileName)
      {
        this.dataStream = stream;
        this.dataOffset = bytesWritten;
        this.path2FlushedData = path2FlushedData;
        this.fileName = fileName;
      }
  
      public void close()
      {
        if (dataStream != null) {
          try {
            dataStream.close();
            bookKeepingPath = new Path(basePath, fileName + BOOK_KEEPING_FILE_OFFSET);
            updateFlushedOffset(bookKeepingPath, dataOffset);
          } catch (IOException ex) {
            logger.warn("not able to close the stream {}", ex.getMessage());
            closeFs();
            throw new RuntimeException(ex);
          }
        }
      }
  
      public void updateOffsets() throws IOException
      {
        updateFlushedOffset(path2FlushedData, dataOffset);
        if (bookKeepingPath != null && fs.exists(bookKeepingPath)) {
          fs.delete(bookKeepingPath, false);
        }
      }
  
    }
  
    private static final Logger logger = LoggerFactory.getLogger(HDFSStorage.class);
  
    @Override
    public void setup(com.datatorrent.api.Context context)
    {
      Configuration conf = new Configuration();
      if (baseDir == null) {
        baseDir = conf.get("hadoop.tmp.dir");
        if (baseDir == null || baseDir.isEmpty()) {
          throw new IllegalArgumentException("baseDir cannot be null.");
        }
      }
      offset = 4;
      skipOffset = -1;
      skipFile = -1;
      int tempRetryCount = 0;
      while (tempRetryCount < retryCount && fs == null) {
        try {
          fs = FileSystem.newInstance(conf);
          tempRetryCount++;
        } catch (Throwable throwable) {
          logger.warn("Not able to get file system ", throwable);
        }
      }
  
      try {
        Path path = new Path(baseDir);
        basePath = new Path(path, id);
        if (fs == null) {
          fs = FileSystem.newInstance(conf);
        }
        if (!fs.exists(path)) {
          closeFs();
          throw new RuntimeException(String.format("baseDir passed (%s) doesn't exist.", baseDir));
        }
        if (!fs.isDirectory(path)) {
          closeFs();
          throw new RuntimeException(String.format("baseDir passed (%s) is not a directory.", baseDir));
        }
        if (!restore) {
          fs.delete(basePath, true);
        }
        if (!fs.exists(basePath) || !fs.isDirectory(basePath)) {
          fs.mkdirs(basePath);
        }
  
        if (blockSize == 0) {
          blockSize = fs.getDefaultBlockSize(new Path(basePath, "tempData"));
        }
        if (blockSize == 0) {
          blockSize = DEFAULT_BLOCK_SIZE;
        }
  
        blockSize = blockSizeMultiple * blockSize;
  
        currentWrittenFile = 0;
        cleanedFileCounter = -1;
        retrievalFile = -1;
        // fileCounterFile = new Path(basePath, IDENTITY_FILE);
        flushedFileCounter = -1;
        // cleanFileCounterFile = new Path(basePath, CLEAN_FILE);
        cleanFileOffsetFile = new Path(basePath, CLEAN_OFFSET_FILE);
        cleanFileOffsetFileTemp = new Path(basePath, CLEAN_OFFSET_FILE_TEMP);
        flushedCounterFile = new Path(basePath, FLUSHED_IDENTITY_FILE);
        flushedCounterFileTemp = new Path(basePath, FLUSHED_IDENTITY_FILE_TEMP);
  
        if (restore) {
          //
          // if (fs.exists(fileCounterFile) && fs.isFile(fileCounterFile)) {
          // //currentWrittenFile = Long.valueOf(new String(readData(fileCounterFile)));
          // }
  
          if (fs.exists(cleanFileOffsetFile) && fs.isFile(cleanFileOffsetFile)) {
            cleanedOffset = readData(cleanFileOffsetFile);
          }
  
          if (fs.exists(flushedCounterFile) && fs.isFile(flushedCounterFile)) {
            String strFlushedFileCounter = new String(readData(flushedCounterFile));
            if (strFlushedFileCounter.isEmpty()) {
              logger.warn("empty flushed file");
            } else {
              flushedFileCounter = Long.valueOf(strFlushedFileCounter);
              flushedFileWriteOffset = getFlushedFileWriteOffset(new Path(basePath, flushedFileCounter + OFFSET_SUFFIX));
              bookKeepingFileOffset = getFlushedFileWriteOffset(
                  new Path(basePath, flushedFileCounter + BOOK_KEEPING_FILE_OFFSET));
            }
  
          }
        }
        fileWriteOffset = flushedFileWriteOffset;
        currentWrittenFile = flushedFileCounter;
        cleanedFileCounter = byteArrayToLong(cleanedOffset, offset) - 1;
        if (currentWrittenFile == -1) {
          ++currentWrittenFile;
          fileWriteOffset = 0;
        }
  
      } catch (IOException io) {
  
        throw new RuntimeException(io);
      }
      storageExecutor = Executors.newSingleThreadExecutor(new NameableThreadFactory("StorageHelper"));
    }
  
    private void closeFs()
    {
      if (fs != null) {
        try {
          fs.close();
          fs = null;
        } catch (IOException e) {
          logger.debug(e.getMessage());
        }
      }
    }
  
    private long getFlushedFileWriteOffset(Path filePath) throws IOException
    {
      if (flushedFileCounter != -1 && fs.exists(filePath)) {
        byte[] flushedFileOffsetByte = readData(filePath);
        if (flushedFileOffsetByte != null && flushedFileOffsetByte.length == 8) {
          return Server.readLong(flushedFileOffsetByte, 0);
        }
      }
      return 0;
    }
  
    @Override
    public void teardown()
    {
      logger.debug("called teardown");
      try {
        if (readStream != null) {
          readStream.close();
        }
        synchronized (HDFSStorage.this) {
          if (nextReadStream != null) {
            nextReadStream.close();
          }
        }
  
      } catch (IOException e) {
        throw new RuntimeException(e);
      } finally {
        closeUnflushedFiles();
        storageExecutor.shutdown();
      }
  
    }
  
    private Runnable getNextStream()
    {
      return new Runnable()
      {
        @Override
        public void run()
        {
          try {
            synchronized (HDFSStorage.this) {
              nextRetrievalFile = retrievalFile + 1;
              if (nextRetrievalFile > flushedFileCounter) {
                nextRetrievalData = null;
                return;
              }
              Path path = new Path(basePath, String.valueOf(nextRetrievalFile));
              Path offsetPath = new Path(basePath, nextRetrievalFile + OFFSET_SUFFIX);
              nextRetrievalData = null;
              nextRetrievalData = readData(path);
              byte[] flushedOffset = readData(offsetPath);
              nextFlushedLong = Server.readLong(flushedOffset, 0);
            }
          } catch (Throwable e) {
            logger.warn("in storage executor ", e);
  
          }
        }
      };
    }
  
  }
  
  