docs: private <IN> void executeDataSink(GenericDataSinkBase<?> sink, int superStep) throws Exception {
  		Operator<?> inputOp = sink.getInput();
  		if (inputOp == null) {
  			throw new InvalidProgramException("The data sink " + sink.getName() + " has no input.");
  		}
  		
  		@SuppressWarnings("unchecked")
  		List<IN> input = (List<IN>) execute(inputOp);
  		
  		@SuppressWarnings("unchecked")
  		GenericDataSinkBase<IN> typedSink = (GenericDataSinkBase<IN>) sink;
  
  		// build the runtime context and compute broadcast variables, if necessary
  		TaskInfo taskInfo = new TaskInfo(typedSink.getName(), 1, 0, 1, 0);
  		RuntimeUDFContext ctx;
  
  		MetricGroup metrics = new UnregisteredMetricsGroup();
  			
  		if (RichOutputFormat.class.isAssignableFrom(typedSink.getUserCodeWrapper().getUserCodeClass())) {
  			ctx = superStep == 0 ? new RuntimeUDFContext(taskInfo, userCodeClassLoader, executionConfig, cachedFiles, accumulators, metrics) :
  					new IterationRuntimeUDFContext(taskInfo, userCodeClassLoader, executionConfig, cachedFiles, accumulators, metrics);
  		} else {
  			ctx = null;
  		}
  
  		typedSink.executeOnCollections(input, ctx, executionConfig);
  	}