class SortingDataInput implements StreamTaskInput<T> {

	@Override
	public InputStatus emitNext(DataOutput<T> output) throws Exception {
		if (sortedInput != null) {
			return emitNextSortedRecord(output);
		}

		InputStatus inputStatus = wrappedInput.emitNext(forwardingDataOutput);
		if (inputStatus == InputStatus.END_OF_INPUT) {
			endSorting();
			return emitNextSortedRecord(output);
		}

		return inputStatus;
	}

	public  SortingDataInput(
			StreamTaskInput<T> wrappedInput,
			TypeSerializer<T> typeSerializer,
			TypeSerializer<K> keySerializer,
			KeySelector<T, K> keySelector,
			MemoryManager memoryManager,
			IOManager ioManager,
			boolean objectReuse,
			double managedMemoryFraction,
			Configuration jobConfiguration,
			AbstractInvokable containingTask);

}