docs: @SuppressWarnings("unchecked")
  	public static <T> Collector<T> initOutputs(AbstractInvokable containingTask, ClassLoader cl, TaskConfig config,
  										List<ChainedDriver<?, ?>> chainedTasksTarget,
  										List<RecordWriter<?>> eventualOutputs,
  										ExecutionConfig executionConfig,
  										Map<String, Accumulator<?,?>> accumulatorMap)
  	throws Exception
  	{
  		final int numOutputs = config.getNumOutputs();
  
  		// check whether we got any chained tasks
  		final int numChained = config.getNumberOfChainedStubs();
  		if (numChained > 0) {
  			// got chained stubs. that means that this one may only have a single forward connection
  			if (numOutputs != 1 || config.getOutputShipStrategy(0) != ShipStrategyType.FORWARD) {
  				throw new RuntimeException("Plan Generation Bug: Found a chained stub that is not connected via an only forward connection.");
  			}
  
  			// instantiate each task
  			@SuppressWarnings("rawtypes")
  			Collector previous = null;
  			for (int i = numChained - 1; i >= 0; --i)
  			{
  				// get the task first
  				final ChainedDriver<?, ?> ct;
  				try {
  					Class<? extends ChainedDriver<?, ?>> ctc = config.getChainedTask(i);
  					ct = ctc.newInstance();
  				}
  				catch (Exception ex) {
  					throw new RuntimeException("Could not instantiate chained task driver.", ex);
  				}
  
  				// get the configuration for the task
  				final TaskConfig chainedStubConf = config.getChainedStubConfig(i);
  				final String taskName = config.getChainedTaskName(i);
  
  				if (i == numChained - 1) {
  					// last in chain, instantiate the output collector for this task
  					previous = getOutputCollector(containingTask, chainedStubConf, cl, eventualOutputs, 0, chainedStubConf.getNumOutputs());
  				}
  
  				ct.setup(chainedStubConf, taskName, previous, containingTask, cl, executionConfig, accumulatorMap);
  				chainedTasksTarget.add(0, ct);
  
  				if (i == numChained - 1) {
  					ct.getIOMetrics().reuseOutputMetricsForTask();
  				}
  
  				previous = ct;
  			}
  			// the collector of the first in the chain is the collector for the task
  			return (Collector<T>) previous;
  		}
  		// else
  
  		// instantiate the output collector the default way from this configuration
  		return getOutputCollector(containingTask , config, cl, eventualOutputs, 0, numOutputs);
  	}