Added unit test for repartitioning when interrupting a current file read.
Added flag that allows the user to make the operator idempotent or not.

   * moved the directory scanning so that it only ocurrs when there are no more pending files in AbstractFSDirectoryInputOperator
   * Overrode emit tuples in AbstractThroughputHashFSDirectoryInputOperator so that scanning is done periodically no matter what.