code: @SuppressWarnings("rawtypes")
  	private static SerFunction<Stream<?>, Stream<?>> adjustTaskFunction(TaskDescriptor taskDescriptor){
  		SerFunction<Stream<?>, Stream<?>> modifiedFunction = taskDescriptor.getFunction();
  		if (taskDescriptor.getId() == 0 && !Entry.class.isAssignableFrom(taskDescriptor.getSourceElementType())){
  			if (Writable.class.isAssignableFrom(taskDescriptor.getSourceElementType())){
  				modifiedFunction = modifiedFunction.compose(stream -> stream.map(s -> ((Entry)s).getValue()));
  			}
  			else {
  				if (taskDescriptor.getInputFormatClass() != null){// only URI based sources will have Input Format
  					ParameterizedType parameterizedType = (ParameterizedType) taskDescriptor.getInputFormatClass().getGenericSuperclass();
  					Type type = parameterizedType.getActualTypeArguments()[1];
  
  					if (Text.class.getName().equals(type.getTypeName())){
  						if (modifiedFunction == null) {
  							modifiedFunction = stream -> stream.map(s -> ((Entry) s).getValue().toString());
  						} else {
  							modifiedFunction = modifiedFunction.compose(stream -> stream.map(s -> ((Entry) s)
  									.getValue().toString()));
  						}
  					}
  					else {
  						//TODO need to design some type of extensible converter to support multiple types of Writable
  						throw new IllegalStateException("Can't determine modified function");
  					}
  				}
  			}
  		}
  		return modifiedFunction;
  	}
docs: This will adjust task function to ensure that it is compatible with Hadoop KV readers and types expected by user.
  For example, reading Text file Tez will produce KV pairs (offset, line), while user is only expected the value.