A modified MoveFileRotation to not rotate zero files

[rangatdt]

MoveFileRotation rotates even 0-sized files. For non-continous streams and especially when microbatching, the number of 0-sized files can become problematic. A modification to the MoveFileRotation to delete 0-sized files instead of moving would be useful. Probably as a configurable option.

E.g.: public void execute(FileSystem fs, Path filePath) throws IOException { Path destPath=new Path(destination,filePath.getName()); long size=fs.getFileStatus(filePath).getLen(); if(size>0) { LOG.info("Moving file {} to {}",filePath,destPath); boolean success=fs.rename(filePath,destPath); if(!success) LOG.warn("Unable to move file {} to {}",filePath,destPath); } else { LOG.info("Deleting 0-sized file {}",filePath); boolean success=fs.delete(filePath); if(!success) LOG.warn("Delete of file {} failed",filePath); } return; }