code: @Override
  	public void close() {
  		try {
  			sortAndReduce();
  		} catch (Exception e) {
  			throw new ExceptionInChainedStubException(this.taskName, e);
  		}
  
  		this.outputCollector.close();
  	}
docs: --------------------------------------------------------------------------------------------