/**
 * 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.lib.io.fs;

import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nonnull;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.apex.malhar.lib.counters.BasicCounters;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.mutable.MutableLong;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.AbstractFileSystem;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.fs.permission.FsPermission;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import com.datatorrent.api.Context;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.Operator;
import com.datatorrent.api.StreamCodec;
import com.datatorrent.api.annotation.OperatorAnnotation;
import com.datatorrent.common.util.BaseOperator;

/**
 * This base implementation for a fault tolerant HDFS output operator,
 * which can handle outputting to multiple files when the output file depends on the tuple.
 * The file a tuple is output to is determined by the getFilePath method. The operator can
 * also output files to rolling mode. In rolling mode by default file names have '.#' appended to the
 * end, where # is an integer. A maximum length for files is specified and whenever the current output
 * file size exceeds the maximum length, output is rolled over to a new file whose name ends in '.#+1'.
 * <br/>
 * <br/>
 * <b>Note:</b> This operator maintains internal state in a map of files names to offsets in the endOffsets
 * field. If the user configures this operator to write to an enormous
 * number of files, there is a risk that the operator will run out of memory. In such a case the
 * user is responsible for maintaining the internal state to prevent the operator from running out
 *
 * BenchMark Results
 * -----------------
 * The operator writes 21 MB/s with the following configuration
 *
 * Container memory size=4G
 * Tuple byte size of 32
 * output to a single file
 * Application window length of 1 second
 *
 * The operator writes 25 MB/s with the following configuration
 *
 * Container memory size=4G
 * Tuple byte size of 32
 * output to a single file
 * Application window length of 30 seconds
 *
 * @displayName FS Writer
 * @category Output
 * @tags fs, file, output operator
 *
 * @param <INPUT> This is the input tuple type.
 *
 * @since 2.0.0
 */
@OperatorAnnotation(checkpointableWithinAppWindow = false)
public abstract class AbstractFileOutputOperator<INPUT> extends BaseOperator implements Operator.CheckpointListener, Operator.CheckpointNotificationListener
{
  private static final Logger LOG = LoggerFactory.getLogger(AbstractFileOutputOperator.class);

  private static final String TMP_EXTENSION = ".tmp";

  private static final String APPEND_TMP_FILE = "_APPENDING";

  private static final int MAX_NUMBER_FILES_IN_TEARDOWN_EXCEPTION = 25;

  /**
   * Size of the copy buffer used to restore files to checkpointed state.
   */
  private static final int COPY_BUFFER_SIZE = 1024;

  /**
   * The default number of max open files.
   */
  public static final int DEFAULT_MAX_OPEN_FILES = 100;

  /**
   * Keyname to rolling file number.
   */
  protected Map<String, MutableInt> openPart;

  /**
   * Keyname to offset of the current rolling file.
   */
  protected Map<String, MutableLong> endOffsets;

  /**
   * Keyname to counts.
   */
  protected Map<String, MutableLong> counts;

  /**
   * Filename to rotation state mapping during a rotation period. Look at {@link #rotationWindows}
   */
  @NotNull
  protected Map<String, RotationState> rotationStates;

  /**
   * The path of the directory to where files are written.
   */
  @NotNull
  protected String filePath;

  /**
   * The total number of bytes written by the operator.
   */
  protected long totalBytesWritten = 0;

  /**
   * The replication level for your output files.
   */
  @Min(0)
  protected int replication = 0;

  /**
   * The maximum number of open files you can have in your streamsCache.
   */
  @Min(1)
  protected int maxOpenFiles = DEFAULT_MAX_OPEN_FILES;

  /**
   * The maximum length in bytes of a rolling file. The default value of this is Long.MAX_VALUE
   */
  @Min(1)
  protected Long maxLength = Long.MAX_VALUE;

  /**
   * The file rotation window interval.
   * The files are rotated periodically after the specified value of windows have ended. If set to 0 this feature is
   * disabled.
   */
  @Min(0)
  protected int rotationWindows = 0;

  /**
   * True if {@link #maxLength} < {@link Long#MAX_VALUE} or {@link #rotationWindows} > 0
   */
  protected transient boolean rollingFile = false;

  /**
   * The file system used to write to.
   */
  protected transient FileSystem fs;
  protected transient FileContext fileContext;

  protected short filePermission = 0777;

  /**
   * This is the cache which holds open file streams.
   */
  protected transient LoadingCache<String, FSFilterStreamContext> streamsCache;

  /**
   * This is the operator context passed at setup.
   */
  protected transient OperatorContext context;

  /**
   * StopWatch tracking the total time the operator has spent writing bytes.
   */
  private transient long totalWritingTime;

  /**
   * File output counters.
   */
  protected final BasicCounters<MutableLong> fileCounters = new BasicCounters<>(MutableLong.class);

  protected StreamCodec<INPUT> streamCodec;

  /**
   * Number of windows since the last rotation
   */
  private int rotationCount;

  /**
   * If a filter stream provider is set it is used to obtain the filter that will be applied to data before it is
   * stored in the file. If it null no filter is applied and data is written as is. Multiple filters can be chained
   * together by using a filter stream chain provider.
   */
  protected FilterStreamProvider filterStreamProvider;

  /**
   * When true this will write to a filename.tmp instead of filename. This is added because at times when the operator
   * gets killed the lease of the last file it was writing to is still open in hdfs. So truncating the file during
   * recovery fails. Ideally this should be false however currently it should remain true otherwise it will lead to
   * namenode instability and crash.
   */
  protected boolean alwaysWriteToTmp = true;

  private final Map<String, String> fileNameToTmpName;
  private final Map<Long, Set<String>> finalizedFiles;
  protected final Map<String, MutableInt> finalizedPart;

  protected long currentWindow;

  /**
   * The stream is expired (closed and evicted from cache) after the specified duration has passed since it was last
   * accessed by a read or write.
   * <p/>
   * https://code.google.com/p/guava-libraries/wiki/CachesExplained <br/>
   * Caches built with CacheBuilder do not perform cleanup and evict values "automatically," or instantly after a
   * value expires, or anything of the sort. Instead, it performs small amounts of maintenance during write
   * operations, or during occasional read operations if writes are rare.<br/>
   * This isn't the most effective way but adds a little bit of optimization.
   */
  private Long expireStreamAfterAccessMillis;
  private final Set<String> filesWithOpenStreams;

  private transient boolean initializeContext;

  /**
   * This input port receives incoming tuples.
   */
  public final transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
  {
    @Override
    public void process(INPUT tuple)
    {
      processTuple(tuple);
    }

    @Override
    public StreamCodec<INPUT> getStreamCodec()
    {
      if (AbstractFileOutputOperator.this.streamCodec == null) {
        return super.getStreamCodec();
      } else {
        return streamCodec;
      }
    }
  };

  private static class RotationState
  {
    boolean notEmpty;
    boolean rotated;
  }

  public AbstractFileOutputOperator()
  {
    endOffsets = Maps.newHashMap();
    counts = Maps.newHashMap();
    openPart = Maps.newHashMap();
    rotationStates = Maps.newHashMap();
    fileNameToTmpName = Maps.newHashMap();
    finalizedFiles = Maps.newTreeMap();
    finalizedPart = Maps.newHashMap();
    filesWithOpenStreams = Sets.newHashSet();
  }

  /**
   * Override this method to change the FileSystem instance that is used by the operator.
   * This method is mainly helpful for unit testing.
   * @return A FileSystem object.
   * @throws IOException
   */
  protected FileSystem getFSInstance() throws IOException
  {
    FileSystem tempFS = FileSystem.newInstance(new Path(filePath).toUri(), new Configuration());

    if (tempFS instanceof LocalFileSystem) {
      tempFS = ((LocalFileSystem)tempFS).getRaw();
    }

    return tempFS;
  }

  @Override
  public void setup(Context.OperatorContext context)
  {
    LOG.debug("setup initiated");
    if (expireStreamAfterAccessMillis == null) {
      expireStreamAfterAccessMillis = (long)(context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) *
        context.getValue(Context.DAGContext.CHECKPOINT_WINDOW_COUNT));
    }
    rollingFile = (maxLength < Long.MAX_VALUE) || (rotationWindows > 0);

    //Getting required file system instance.
    try {
      fs = getFSInstance();
    } catch (IOException ex) {
      throw new RuntimeException(ex);
    }

    if (replication <= 0) {
      replication = fs.getDefaultReplication(new Path(filePath));
    }

    LOG.debug("FS class {}", fs.getClass());

    //building cache
    RemovalListener<String, FSFilterStreamContext> removalListener = createCacheRemoveListener();
    CacheLoader<String, FSFilterStreamContext> loader = createCacheLoader();
    streamsCache = CacheBuilder.newBuilder().maximumSize(maxOpenFiles).expireAfterAccess(expireStreamAfterAccessMillis,
      TimeUnit.MILLISECONDS).removalListener(removalListener).build(loader);

    LOG.debug("File system class: {}", fs.getClass());
    LOG.debug("end-offsets {}", endOffsets);

    try {
      //Restore the files in case they were corrupted and the operator was re-deployed.
      Path writerPath = new Path(filePath);
      if (fs.exists(writerPath)) {
        for (String seenFileName : endOffsets.keySet()) {
          String seenFileNamePart = getPartFileNamePri(seenFileName);
          LOG.debug("seenFileNamePart: {}", seenFileNamePart);

          Path activeFilePath;
          if (alwaysWriteToTmp) {
            String tmpFileName = fileNameToTmpName.get(seenFileNamePart);
            activeFilePath = new Path(filePath + Path.SEPARATOR + tmpFileName);
          } else {
            activeFilePath = new Path(filePath + Path.SEPARATOR + seenFileNamePart);
          }

          if (fs.exists(activeFilePath)) {
            recoverFile(seenFileName, seenFileNamePart, activeFilePath);
          }
        }
      }

      if (rollingFile) {
        //delete the left over future rolling files produced from the previous crashed instance of this operator.
        for (String seenFileName : endOffsets.keySet()) {
          try {
            Integer fileOpenPart = this.openPart.get(seenFileName).getValue();
            int nextPart = fileOpenPart + 1;
            String seenPartFileName;
            while (true) {
              seenPartFileName = getPartFileName(seenFileName, nextPart);
              Path activePath = null;
              if (alwaysWriteToTmp) {
                String tmpFileName = fileNameToTmpName.get(seenPartFileName);
                if (tmpFileName != null) {
                  activePath = new Path(filePath + Path.SEPARATOR + tmpFileName);
                }
              } else {
                activePath = new Path(filePath + Path.SEPARATOR + seenPartFileName);
              }
              if (activePath == null || !fs.exists(activePath)) {
                break;
              }

              fs.delete(activePath, true);
              nextPart++;
            }

            seenPartFileName = getPartFileName(seenFileName, fileOpenPart);
            Path activePath = null;
            if (alwaysWriteToTmp) {
              String tmpFileName = fileNameToTmpName.get(seenPartFileName);
              if (tmpFileName != null) {
                activePath = new Path(filePath + Path.SEPARATOR + fileNameToTmpName.get(seenPartFileName));
              }
            } else {
              activePath = new Path(filePath + Path.SEPARATOR + seenPartFileName);
            }

            if (activePath != null && fs.exists(activePath) && fs.getFileStatus(activePath).getLen() > maxLength) {
              //Handle the case when restoring to a checkpoint where the current rolling file
              //already has a length greater than max length.
              LOG.debug("rotating file at setup.");
              rotate(seenFileName);
            }
          } catch (IOException | ExecutionException e) {
            throw new RuntimeException(e);
          }
        }
      }
      LOG.debug("setup completed");
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    this.context = context;

    fileCounters.setCounter(Counters.TOTAL_BYTES_WRITTEN, new MutableLong());
    fileCounters.setCounter(Counters.TOTAL_TIME_WRITING_MILLISECONDS, new MutableLong());
  }

  /**
   * Recovers a file which exists on the disk. If the length of the file is not same as the
   * length which the operator remembers then the file is truncated. <br/>
   * When always writing to a temporary file, then a file is restored even when the length is same as what the
   * operator remembers however this is done only for files which had open streams that weren't closed before
   * failure.
   *
   * @param filename     name of the actual file.
   * @param partFileName name of the part file. When not rolling this is same as filename; otherwise this is the
   *                     latest open part file name.
   * @param filepath     path of the file. When always writing to temp file, this is the path of the temp file;
   *                     otherwise path of the actual file.
   * @throws IOException
   */
  private void recoverFile(String filename, String partFileName, Path filepath) throws IOException
  {
    LOG.debug("path exists {}", filepath);
    long offset = endOffsets.get(filename).longValue();
    FSDataInputStream inputStream = fs.open(filepath);
    FileStatus status = fs.getFileStatus(filepath);

    if (status.getLen() != offset) {
      LOG.info("path corrupted {} {} {}", filepath, offset, status.getLen());
      byte[] buffer = new byte[COPY_BUFFER_SIZE];
      String recoveryFileName = partFileName + '.' + System.currentTimeMillis() + TMP_EXTENSION;
      Path recoveryFilePath = new Path(filePath + Path.SEPARATOR + recoveryFileName);
      FSDataOutputStream fsOutput = openStream(recoveryFilePath, false);

      while (inputStream.getPos() < offset) {
        long remainingBytes = offset - inputStream.getPos();
        int bytesToWrite = remainingBytes < COPY_BUFFER_SIZE ? (int)remainingBytes : COPY_BUFFER_SIZE;
        inputStream.read(buffer);
        fsOutput.write(buffer, 0, bytesToWrite);
      }

      flush(fsOutput);
      fsOutput.close();
      inputStream.close();

      LOG.debug("active {} recovery {} ", filepath, recoveryFilePath);

      if (alwaysWriteToTmp) {
        //recovery file is used as the new tmp file and we cannot delete the old tmp file because when the operator
        //is restored to an earlier check-pointed window, it will look for an older tmp.
        fileNameToTmpName.put(partFileName, recoveryFileName);
      } else {
        LOG.debug("recovery path {} actual path {} ", recoveryFilePath, status.getPath());
        rename(recoveryFilePath, status.getPath());
      }
    } else {
      if (alwaysWriteToTmp && filesWithOpenStreams.contains(filename)) {
        String currentTmp = partFileName + '.' + System.currentTimeMillis() + TMP_EXTENSION;
        FSDataOutputStream outputStream = openStream(new Path(filePath + Path.SEPARATOR + currentTmp), false);
        IOUtils.copy(inputStream, outputStream);
        streamsCache.put(filename, new FSFilterStreamContext(outputStream));
        fileNameToTmpName.put(partFileName, currentTmp);
      }
      inputStream.close();
    }
  }

  /**
   * Creates the {@link CacheLoader} for loading an output stream when it is not present in the cache.
   * @return cache loader
   */
  private CacheLoader<String, FSFilterStreamContext> createCacheLoader()
  {
    return new CacheLoader<String, FSFilterStreamContext>()
    {
      @Override
      public FSFilterStreamContext load(@Nonnull String filename)
      {
        if (rollingFile) {
          RotationState state = getRotationState(filename);
          if (rollingFile && state.rotated) {
            openPart.get(filename).add(1);
            state.rotated = false;
            MutableLong offset = endOffsets.get(filename);
            offset.setValue(0);
          }
        }

        String partFileName = getPartFileNamePri(filename);
        Path originalFilePath = new Path(filePath + Path.SEPARATOR + partFileName);

        Path activeFilePath;
        if (!alwaysWriteToTmp) {
          activeFilePath = originalFilePath;
        } else {
          //MLHR-1776 : writing to tmp file
          String tmpFileName = fileNameToTmpName.get(partFileName);
          if (tmpFileName == null) {
            tmpFileName = partFileName + '.' + System.currentTimeMillis() + TMP_EXTENSION;
            fileNameToTmpName.put(partFileName, tmpFileName);
          }
          activeFilePath = new Path(filePath + Path.SEPARATOR + tmpFileName);
        }

        FSDataOutputStream fsOutput;

        boolean sawThisFileBefore = endOffsets.containsKey(filename);

        try {
          if (fs.exists(originalFilePath) || (alwaysWriteToTmp && fs.exists(activeFilePath))) {
            if (sawThisFileBefore) {
              FileStatus fileStatus = fs.getFileStatus(activeFilePath);
              MutableLong endOffset = endOffsets.get(filename);

              if (endOffset != null) {
                endOffset.setValue(fileStatus.getLen());
              } else {
                endOffsets.put(filename, new MutableLong(fileStatus.getLen()));
              }

              fsOutput = openStream(activeFilePath, true);
              LOG.debug("appending to {}", activeFilePath);
            } else {
              //We never saw this file before and we don't want to append
              //If the file is rolling we need to delete all its parts.
              if (rollingFile) {
                int part = 0;

                while (true) {
                  Path seenPartFilePath = new Path(filePath + Path.SEPARATOR + getPartFileName(filename, part));
                  if (!fs.exists(seenPartFilePath)) {
                    break;
                  }

                  fs.delete(seenPartFilePath, true);
                  part = part + 1;
                }

                fsOutput = openStream(activeFilePath, false);
              } else {
                //Not rolling is easy, just delete the file and create it again.
                fs.delete(activeFilePath, true);
                if (alwaysWriteToTmp) {
                  //we need to delete original file if that exists
                  if (fs.exists(originalFilePath)) {
                    fs.delete(originalFilePath, true);
                  }
                }
                fsOutput = openStream(activeFilePath, false);
              }
            }
          } else {
            fsOutput = openStream(activeFilePath, false);
          }
          filesWithOpenStreams.add(filename);

          LOG.info("opened {}, active {}", partFileName, activeFilePath);
          return new FSFilterStreamContext(fsOutput);
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
      }
    };
  }

  /**
   * Creates the removal listener which is attached to the cache.
   *
   * @return cache entry removal listener.
   */
  private RemovalListener<String, FSFilterStreamContext> createCacheRemoveListener()
  {
    //When an entry is removed from the cache, removal listener is notified and it closes the output stream.
    return new RemovalListener<String, FSFilterStreamContext>()
    {
      @Override
      public void onRemoval(@Nonnull RemovalNotification<String, FSFilterStreamContext> notification)
      {
        FSFilterStreamContext streamContext = notification.getValue();
        if (streamContext != null) {
          try {
            String filename = notification.getKey();
            String partFileName = getPartFileNamePri(filename);

            LOG.info("closing {}", partFileName);
            long start = System.currentTimeMillis();

            closeStream(streamContext);
            filesWithOpenStreams.remove(filename);

            totalWritingTime += System.currentTimeMillis() - start;
          } catch (IOException e) {
            LOG.error("removing {}", notification.getValue(), e);
            throw new RuntimeException(e);
          }
        }
      }
    };
  }

  /**
   * Opens the stream for the specified file path in either append mode or create mode.
   *
   * @param filepath this is the path of either the actual file or the corresponding temporary file.
   * @param append   true for opening the file in append mode; false otherwise.
   * @return output stream.
   * @throws IOException
   */
  protected FSDataOutputStream openStream(Path filepath, boolean append) throws IOException
  {
    FSDataOutputStream fsOutput;
    if (append) {
      fsOutput = openStreamInAppendMode(filepath);
    } else {
      fsOutput = fs.create(filepath, (short)replication);
      fs.setPermission(filepath, FsPermission.createImmutable(filePermission));
    }
    return fsOutput;
  }

  /**
   * Opens the stream for the given file path in append mode. Catch the exception if the FS doesnt support
   * append operation and calls the openStreamForNonAppendFS().
   * @param filepath given file path
   * @return output stream
   */
  protected FSDataOutputStream openStreamInAppendMode(Path filepath)
  {
    FSDataOutputStream fsOutput = null;
    try {
      fsOutput = fs.append(filepath);
    } catch (IOException e) {
      if (e.getMessage().equals("Not supported")) {
        fsOutput = openStreamForNonAppendFS(filepath);
      }
    }
    return fsOutput;
  }

  /**
   * Opens the stream for the given file path for the file systems which are not supported append operation.
   * @param filepath given file path
   * @return output stream
   */
  protected FSDataOutputStream openStreamForNonAppendFS(Path filepath)
  {
    try {
      Path appendTmpFile = new Path(filepath + APPEND_TMP_FILE);
      fs.rename(filepath, appendTmpFile);
      FSDataInputStream fsIn = fs.open(appendTmpFile);
      FSDataOutputStream fsOut = fs.create(filepath);
      IOUtils.copy(fsIn, fsOut);
      flush(fsOut);
      fsIn.close();
      fs.delete(appendTmpFile);
      return fsOut;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * Closes the stream which has been removed from the cache.
   *
   * @param streamContext stream context which is removed from the cache.
   * @throws IOException
   */
  protected void closeStream(FSFilterStreamContext streamContext) throws IOException
  {
    streamContext.close();
  }

  /**
   * Renames source path to destination atomically. This relies on the FileContext api. If
   * the underlying filesystem doesn't have an {@link AbstractFileSystem} then this should be overridden.
   *
   * @param source      source path
   * @param destination destination path
   * @throws IOException
   */
  protected void rename(Path source, Path destination) throws IOException
  {
    if (fileContext == null) {
      fileContext = FileContext.getFileContext(fs.getUri());
    }
    fileContext.rename(source, destination, Options.Rename.OVERWRITE);
  }

  /**
   * Requests a file to be finalized. When it is writing to a rolling file, this will
   * request for finalizing the current open part and all the prev parts which weren't requested yet.
   *
   * @param fileName name of the file; part file name in case of rotation.
   * @throws IOException
   */
  protected void requestFinalize(String fileName)
  {
    Set<String> filesPerWindow = finalizedFiles.get(currentWindow);
    if (filesPerWindow == null) {
      filesPerWindow = Sets.newHashSet();
      finalizedFiles.put(currentWindow, filesPerWindow);
    }
    if (rollingFile) {

      MutableInt part = finalizedPart.get(fileName);
      if (part == null) {
        part = new MutableInt(-1);
        finalizedPart.put(fileName, part);
      }
      MutableInt currentOpenPart = openPart.get(fileName);

      for (int x = part.getValue() + 1; x <= currentOpenPart.getValue(); x++) {
        String prevPartNotFinalized = getPartFileName(fileName, x);
        LOG.debug("request finalize {}", prevPartNotFinalized);
        filesPerWindow.add(prevPartNotFinalized);
      }
      fileName = getPartFileNamePri(fileName);
      part.setValue(currentOpenPart.getValue());
    }
    filesPerWindow.add(fileName);
  }

  @Override
  public void teardown()
  {
    List<String> fileNames = new ArrayList<String>();
    int numberOfFailures = 0;
    IOException savedException = null;

    //Close all the streams you can
    Map<String, FSFilterStreamContext> openStreams = streamsCache.asMap();
    for (String seenFileName : openStreams.keySet()) {
      FSFilterStreamContext fsFilterStreamContext = openStreams.get(seenFileName);
      try {
        long start = System.currentTimeMillis();
        closeStream(fsFilterStreamContext);
        filesWithOpenStreams.remove(seenFileName);
        totalWritingTime += System.currentTimeMillis() - start;
      } catch (IOException ex) {
        //Count number of failures
        numberOfFailures++;
        //Add names of first N failed files to list
        if (fileNames.size() < MAX_NUMBER_FILES_IN_TEARDOWN_EXCEPTION) {
          fileNames.add(seenFileName);
          //save exception
          savedException = ex;
        }
      }
    }

    //Try to close the file system
    boolean fsFailed = false;

    try {
      fs.close();
    } catch (IOException ex) {
      //Closing file system failed
      savedException = ex;
      fsFailed = true;
    }

    //There was a failure closing resources
    if (savedException != null) {
      String errorMessage = "";

      //File system failed to close
      if (fsFailed) {
        errorMessage += "Closing the fileSystem failed. ";
      }

      //Print names of atmost first N files that failed to close
      if (!fileNames.isEmpty()) {
        errorMessage += "The following files failed closing: ";
      }

      for (String seenFileName: fileNames) {
        errorMessage += seenFileName + ", ";
      }

      if (numberOfFailures > MAX_NUMBER_FILES_IN_TEARDOWN_EXCEPTION) {
        errorMessage += (numberOfFailures - MAX_NUMBER_FILES_IN_TEARDOWN_EXCEPTION) +
                        " more files failed.";
      }

      //Fail
      throw new RuntimeException(errorMessage, savedException);
    }

    long currentTimeStamp = System.currentTimeMillis();
  }

  /**
   * This method processes received tuples.
   * Tuples are written out to the appropriate files as determined by the getFileName method.
   * If the output port is connected incoming tuples are also converted and emitted on the appropriate output port.
   * @param tuple An incoming tuple which needs to be processed.
   */
  protected void processTuple(INPUT tuple)
  {
    String fileName = getFileName(tuple);

    if (Strings.isNullOrEmpty(fileName)) {
      return;
    }

    try {
      FilterOutputStream fsOutput = streamsCache.get(fileName).getFilterStream();
      byte[] tupleBytes = getBytesForTuple(tuple);
      long start = System.currentTimeMillis();
      fsOutput.write(tupleBytes);
      totalWritingTime += System.currentTimeMillis() - start;
      totalBytesWritten += tupleBytes.length;
      MutableLong currentOffset = endOffsets.get(fileName);

      if (currentOffset == null) {
        currentOffset = new MutableLong(0);
        endOffsets.put(fileName, currentOffset);
      }

      currentOffset.add(tupleBytes.length);

      if (rotationWindows > 0) {
        getRotationState(fileName).notEmpty = true;
      }

      if (rollingFile && currentOffset.longValue() > maxLength) {
        LOG.debug("Rotating file {} {} {}", fileName, openPart.get(fileName), currentOffset.longValue());
        rotate(fileName);
      }

      MutableLong count = counts.get(fileName);
      if (count == null) {
        count = new MutableLong(0);
        counts.put(fileName, count);
      }

      count.add(1);
    } catch (IOException | ExecutionException ex) {
      throw new RuntimeException(ex);
    }
  }

  /**
   * This method rolls over to the next files.
   * @param fileName The file that you are rolling.
   * @throws IllegalArgumentException
   * @throws IOException
   * @throws ExecutionException
   */
  protected void rotate(String fileName) throws IllegalArgumentException, IOException, ExecutionException
  {
    if (!this.getRotationState(fileName).rotated) {
      requestFinalize(fileName);
      counts.remove(fileName);
      streamsCache.invalidate(fileName);
      MutableInt mi = openPart.get(fileName);
      LOG.debug("Part file rotated {} : {}", fileName, mi.getValue());

      //TODO: remove this as rotateHook is deprecated.
      String partFileName = getPartFileName(fileName, mi.getValue());
      rotateHook(partFileName);

      getRotationState(fileName).rotated = true;
    }
  }

  private RotationState getRotationState(String fileName)
  {
    RotationState rotationState = rotationStates.get(fileName);
    if (rotationState == null) {
      rotationState = new RotationState();
      rotationStates.put(fileName, rotationState);
    }
    return rotationState;
  }

  /**
   * This hook is called after a rolling file part has filled up and is closed. The hook is passed
   * the name of the file part that has just completed closed.
   * @param finishedFile The name of the file part that has just completed and closed.
   */
  @Deprecated
  protected void rotateHook(@SuppressWarnings("unused") String finishedFile)
  {
    //Do nothing by default
  }

  /**
   * This method is used to force buffers to be flushed at the end of the window.
   * flush must be used on a local file system, so an if statement checks to
   * make sure that hflush is used on local file systems.
   * @param fsOutput      output stream
   * @throws IOException
   */
  protected void flush(FSDataOutputStream fsOutput) throws IOException
  {
    if (fs instanceof LocalFileSystem ||
        fs instanceof RawLocalFileSystem) {
      fsOutput.flush();
    } else {
      fsOutput.hflush();
    }
  }

  /**
   * Gets the current rolling file name.
   * @param fileName The base name of the files you are rolling over.
   * @return The name of the current rolling file.
   */
  protected String getPartFileNamePri(String fileName)
  {
    if (!rollingFile) {
      return fileName;
    }

    MutableInt part = openPart.get(fileName);

    if (part == null) {
      part = new MutableInt(0);
      openPart.put(fileName, part);
      LOG.debug("First file part number {}", part);
    }

    return getPartFileName(fileName,
                           part.intValue());
  }

  /**
   * This method constructs the next rolling files name based on the
   * base file name and the number in the sequence of rolling files.
   * @param fileName The base name of the file.
   * @param part The part number of the rolling file.
   * @return The rolling file name.
   */
  protected String getPartFileName(String fileName, int part)
  {
    return fileName + "." + part;
  }

  @Override
  public void beginWindow(long windowId)
  {
    // All the filter state needs to be flushed to the disk. Not all filters allow a flush option, so the filters have
    // to be closed and reopened. If no filter being is being used then it is a essentially a noop as the underlying
    // FSDataOutputStream is not being closed in this operation.
    if (initializeContext) {
      try {
        Map<String, FSFilterStreamContext> openStreams = streamsCache.asMap();
        for (FSFilterStreamContext streamContext : openStreams.values()) {
          streamContext.initializeContext();
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      initializeContext = false;
    }
    currentWindow = windowId;
  }

  @Override
  public void endWindow()
  {
    if (rotationWindows > 0) {
      if (++rotationCount == rotationWindows) {
        rotationCount = 0;
        // Rotate the files
        Iterator<Map.Entry<String, MutableInt>> iterator = openPart.entrySet().iterator();
        while (iterator.hasNext()) {
          String filename = iterator.next().getKey();
          // Rotate the file if the following conditions are met
          // 1. The file is not already rotated during this period for other reasons such as max length is reached
          //     or rotate was explicitly called externally
          // 2. The file is not empty
          RotationState rotationState = rotationStates.get(filename);
          boolean rotate = false;
          if (rotationState != null) {
            rotate = !rotationState.rotated && rotationState.notEmpty;
            rotationState.notEmpty = false;
          }
          if (rotate) {
            try {
              rotate(filename);
            } catch (IOException | ExecutionException e) {
              throw new RuntimeException(e);
            }
          }
        }
      }
    }

    fileCounters.getCounter(Counters.TOTAL_TIME_WRITING_MILLISECONDS).setValue(totalWritingTime);
    fileCounters.getCounter(Counters.TOTAL_BYTES_WRITTEN).setValue(totalBytesWritten);
    context.setCounters(fileCounters);
  }

  /**
   * This method determines the file that a received tuple is written out to.
   * @param tuple The tuple which can be used to determine the output file name.
   * @return The name of the file this tuple will be written out to.
   */
  protected abstract String getFileName(INPUT tuple);

  /**
   * This method converts a tuple into bytes, so that it can be written out to a filesystem.
   * @param tuple A received tuple to be converted into bytes.
   * @return The received tuple in byte form.
   */
  protected abstract byte[] getBytesForTuple(INPUT tuple);

  /**
   * Sets the path of the working directory where files are being written.
   * @param dir The path of the working directory where files are being written.
   */
  public void setFilePath(@Nonnull String dir)
  {
    this.filePath = dir;
  }

  /**
   * Returns the path of the working directory where files are being written.
   * @return The path of the working directory where files are being written.
   */
  public String getFilePath()
  {
    return this.filePath;
  }

  /**
   * Sets the maximum length of a an output file in bytes. By default this is Long.MAX_VALUE,
   * if this is not Long.MAX_VALUE then the output operator is in rolling mode.
   * @param maxLength The maximum length of an output file in bytes, when in rolling mode.
   */
  public void setMaxLength(long maxLength)
  {
    this.maxLength = maxLength;
  }

  /**
   * Sets the maximum length of a an output file in bytes, when in rolling mode. By default
   * this is null, if this is not null then the operator is in rolling mode.
   * @return The maximum length of an output file in bytes, when in rolling mode.
   */
  public long getMaxLength()
  {
    return maxLength;
  }

  /**
   * Gets the file rotation window interval.
   * The files are rotated periodically after the specified number of windows have ended.
   * @return The number of windows
   */
  public int getRotationWindows()
  {
    return rotationWindows;
  }

  /**
   * Sets the file rotation window interval.
   * The files are rotated periodically after the specified number of windows have ended.
   * @param rotationWindows The number of windows
   */
  public void setRotationWindows(int rotationWindows)
  {
    this.rotationWindows = rotationWindows;
  }

  /**
   * Set the maximum number of files which can be written to at a time.
   * @param maxOpenFiles The maximum number of files which can be written to at a time.
   */
  public void setMaxOpenFiles(int maxOpenFiles)
  {
    this.maxOpenFiles = maxOpenFiles;
  }

  /**
   * Get the maximum number of files which can be open.
   * @return The maximum number of files which can be open.
   */
  public int getMaxOpenFiles()
  {
    return this.maxOpenFiles;
  }

   /**
   * Get the permission on the file which is being written.
   * @return filePermission
   */
  public short getFilePermission()
  {
    return filePermission;
  }

  /**
   * Set the permission on the file which is being written.
   * @param filePermission
   */
  public void setFilePermission(short filePermission)
  {
    this.filePermission = filePermission;
  }

  /**
   * Get the filter stream provider
   * @return The filter stream provider.
   */
  public FilterStreamProvider getFilterStreamProvider()
  {
    return filterStreamProvider;
  }

  /**
   * Set the filter stream provider. When a non-null provider is specified it will be used to supply the filter that
   * will be  applied to data before it is stored in the file.
   * @param filterStreamProvider The filter stream provider
   */
  public void setFilterStreamProvider(FilterStreamProvider filterStreamProvider)
  {
    this.filterStreamProvider = filterStreamProvider;
  }

  public static enum Counters
  {
    /**
     * An enum for counters representing the total number of bytes written
     * by the operator.
     */
    TOTAL_BYTES_WRITTEN,

    /**
     * An enum for counters representing the total time the operator has
     * been operational for.
     */
    TOTAL_TIME_WRITING_MILLISECONDS
  }

  protected class FSFilterStreamContext implements FilterStreamContext<FilterOutputStream>
  {

    private FSDataOutputStream outputStream;

    private FilterStreamContext filterContext;
    private NonCloseableFilterOutputStream outputWrapper;

    public FSFilterStreamContext(FSDataOutputStream outputStream) throws IOException
    {
      this.outputStream = outputStream;
      outputWrapper = new NonCloseableFilterOutputStream(outputStream);
      //resetFilter();
      initializeContext();
    }

    @Override
    public FilterOutputStream getFilterStream()
    {
      if (filterContext != null) {
        return filterContext.getFilterStream();
      }
      return outputStream;
    }

    @SuppressWarnings("unchecked")
    @Override
    public void finalizeContext() throws IOException
    {
      if (filterContext != null) {
        filterContext.finalizeContext();
        outputWrapper.flush();
      }
      outputStream.hflush();
      if (filterStreamProvider != null) {
        filterStreamProvider.reclaimFilterStreamContext(filterContext);
      }
    }

    @SuppressWarnings("unchecked")
    public void initializeContext() throws IOException
    {
      if (filterStreamProvider != null) {
        filterContext = filterStreamProvider.getFilterStreamContext(outputWrapper);
      }
    }

    public void close() throws IOException
    {
      //finalizeContext();
      if (filterContext != null) {
        filterContext.getFilterStream().close();
      }
      outputStream.close();
    }

  }

  private static class NonCloseableFilterOutputStream extends FilterOutputStream
  {
    public NonCloseableFilterOutputStream(OutputStream out)
    {
      super(out);
    }

    @Override
    public void close() throws IOException
    {
    }
  }

  @Override
  public void beforeCheckpoint(long l)
  {
    try {
      Map<String, FSFilterStreamContext> openStreams = streamsCache.asMap();
      for (FSFilterStreamContext streamContext: openStreams.values()) {
        long start = System.currentTimeMillis();
        streamContext.finalizeContext();
        totalWritingTime += System.currentTimeMillis() - start;
        // Re-initialize context when next window starts after checkpoint
        initializeContext = true;
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public void checkpointed(long l)
  {
  }

  @Override
  public void committed(long l)
  {
    if (alwaysWriteToTmp) {
      Iterator<Map.Entry<Long, Set<String>>> finalizedFilesIter = finalizedFiles.entrySet().iterator();
      try {

        while (finalizedFilesIter.hasNext()) {
          Map.Entry<Long, Set<String>> filesPerWindow = finalizedFilesIter.next();
          if (filesPerWindow.getKey() > l) {
            break;
          }
          for (String file : filesPerWindow.getValue()) {
            finalizeFile(file);
          }
          finalizedFilesIter.remove();
        }
      } catch (IOException e) {
        throw new RuntimeException("failed to commit", e);
      }
    }
  }

  /**
   * Finalizing a file means that the same file will never be open again.
   *
   * @param fileName name of the file to finalize
   */
  protected void finalizeFile(String fileName) throws IOException
  {
    String tmpFileName = fileNameToTmpName.get(fileName);
    Path srcPath = new Path(filePath + Path.SEPARATOR + tmpFileName);
    Path destPath = new Path(filePath + Path.SEPARATOR + fileName);

    if (!fs.exists(destPath)) {
      LOG.debug("rename from tmp {} actual {} ", tmpFileName, fileName);
      rename(srcPath, destPath);
    } else if (fs.exists(srcPath)) {
      /*if the destination and src both exists that means there was a failure between file rename and clearing the
      endOffset so we just delete the tmp file*/
      LOG.debug("deleting tmp {}", tmpFileName);
      fs.delete(srcPath, true);
    }
    endOffsets.remove(fileName);
    fileNameToTmpName.remove(fileName);

    //when writing to tmp files there can be vagrant tmp files which we have to clean
    FileStatus[] statuses = fs.listStatus(destPath.getParent());
    for (FileStatus status : statuses) {
      String statusName = status.getPath().getName();
      if (statusName.endsWith(TMP_EXTENSION) && statusName.startsWith(destPath.getName())) {
        //a tmp file has tmp extension always preceded by timestamp
        String actualFileName = statusName.substring(0, statusName.lastIndexOf('.', statusName.lastIndexOf('.') - 1));
        if (fileName.equals(actualFileName)) {
          LOG.debug("deleting stray file {}", statusName);
          fs.delete(status.getPath(), true);
        }
      }
    }
  }

  /**
   * @return true if writing to a tmp file rather than the actual file. false otherwise.
   */
  public boolean isAlwaysWriteToTmp()
  {
    return alwaysWriteToTmp;
  }

  /**
   * This controls if data is always written to a tmp file rather than the actual file. Tmp files are renamed to actual
   * files when files are finalized.
   * @param alwaysWriteToTmp true if write to a tmp file; false otherwise.
   */
  public void setAlwaysWriteToTmp(boolean alwaysWriteToTmp)
  {
    this.alwaysWriteToTmp = alwaysWriteToTmp;
  }

  @VisibleForTesting
  protected Map<String, String> getFileNameToTmpName()
  {
    return fileNameToTmpName;
  }

  @VisibleForTesting
  protected Map<Long, Set<String>> getFinalizedFiles()
  {
    return finalizedFiles;
  }

  public Long getExpireStreamAfterAccessMillis()
  {
    return expireStreamAfterAccessMillis;
  }

  /**
   * Sets the duration after which the stream is expired (closed and removed from the cache) since it was last accessed
   * by a read or write.
   *
   * @param millis time in millis.
   */
  public void setExpireStreamAfterAccessMillis(Long millis)
  {
    this.expireStreamAfterAccessMillis = millis;
  }

/**
   * Return the filter to use. If this method returns a filter the filter is applied to data before the data is stored
   * in the file. If it returns null no filter is applied and data is written as is. Override this method to provide
   * the filter implementation. Multiple filters can be chained together to return a chain filter.
   *
   * @param outputStream
   * @return
   */
  /*
  protected FilterStreamContext getStreamContext(OutputStream outputStream) throws IOException
  {
    return null;
  }
  */

}