docs: public static <T> Collector<T> getOutputCollector(AbstractInvokable task, TaskConfig config, ClassLoader cl,
  			List<RecordWriter<?>> eventualOutputs, int outputOffset, int numOutputs) throws Exception
  	{
  		if (numOutputs == 0) {
  			return null;
  		}
  
  		// get the factory for the serializer
  		final TypeSerializerFactory<T> serializerFactory = config.getOutputSerializer(cl);
  		final List<RecordWriter<SerializationDelegate<T>>> writers = new ArrayList<>(numOutputs);
  
  		// create a writer for each output
  		for (int i = 0; i < numOutputs; i++)
  		{
  			// create the OutputEmitter from output ship strategy
  			final ShipStrategyType strategy = config.getOutputShipStrategy(i);
  			final int indexInSubtaskGroup = task.getIndexInSubtaskGroup();
  			final TypeComparatorFactory<T> compFactory = config.getOutputComparator(i, cl);
  
  			final ChannelSelector<SerializationDelegate<T>> oe;
  			if (compFactory == null) {
  				oe = new OutputEmitter<T>(strategy, indexInSubtaskGroup);
  			}
  			else {
  				final DataDistribution dataDist = config.getOutputDataDistribution(i, cl);
  				final Partitioner<?> partitioner = config.getOutputPartitioner(i, cl);
  
  				final TypeComparator<T> comparator = compFactory.createComparator();
  				oe = new OutputEmitter<T>(strategy, indexInSubtaskGroup, comparator, partitioner, dataDist);
  			}
  
  			final RecordWriter<SerializationDelegate<T>> recordWriter = new RecordWriterBuilder()
  				.setChannelSelector(oe)
  				.setTaskName(task.getEnvironment().getTaskInfo().getTaskName())
  				.build(task.getEnvironment().getWriter(outputOffset + i));
  
  			recordWriter.setMetricGroup(task.getEnvironment().getMetricGroup().getIOMetricGroup());
  
  			writers.add(recordWriter);
  		}
  		if (eventualOutputs != null) {
  			eventualOutputs.addAll(writers);
  		}
  		return new OutputCollector<T>(writers, serializerFactory.getSerializer());
  	}