code: public int cleanUp(String dir, boolean fetchAll) {
          if (!dir.endsWith("/")) {
              dir += "/";
          }
  
          int count = 0;
  
          try {
              FileSystem fs = writer.getFS();
              FileStatus[] files = fs.listStatus(new Path(dir));
              for (FileStatus file: files) {
                  if (file.getLen() > 0) {
                      String fileName = file.getPath().getName();
                      String fileExt = getFileExt(fileName);
                      if (fileExt != null && fileExt.equals(done)) {
                          notice.send(dir + fileName);
                          ++count;
                      } else if (fileExt != null) {
                          long lastPeriod =
                                  new DateTime().minus(rotationPeriod).minus(rotationPeriod).getMillis();
                          if (file.getModificationTime() < lastPeriod) {
                              ++errorClosedFiles;
                              DynamicCounter.increment("closedFileError");
                              log.error(dir + fileName + " is not closed properly!!!");
                              String doneFile = fileName.replace(fileExt, done);
                              writer.setDone(dir + fileName, dir + doneFile);
                              notice.send(dir + doneFile);
                              ++count;
                          } else if (fetchAll) {
                              ++count;
                          }
                      }
                  }
              }
          } catch (Exception e) {
              log.error("Exception while on cleanUp: " + e.getMessage(), e);
              return Integer.MAX_VALUE; // return non-zero value
          }
  
          return count;
      }
docs: List all files under the directory. If the file is marked as done, the
  notice for that file would be sent. Otherwise, it checks the file
  is not closed properly, the file is marked as done and the notice
  would be sent. That file would cause EOFException when reading.
  
  @param dir
  @return the number of files found in the directory