code: private <T> Collection<Integer> transformSink(SinkTransformation<T> sink) {
  
  		Collection<Integer> inputIds = transform(sink.getInput());
  
  		String slotSharingGroup = determineSlotSharingGroup(sink.getSlotSharingGroup(), inputIds);
  
  		streamGraph.addSink(sink.getId(),
  				slotSharingGroup,
  				sink.getCoLocationGroupKey(),
  				sink.getOperator(),
  				sink.getInput().getOutputType(),
  				null,
  				"Sink: " + sink.getName());
  
  		streamGraph.setParallelism(sink.getId(), sink.getParallelism());
  		streamGraph.setMaxParallelism(sink.getId(), sink.getMaxParallelism());
  
  		for (Integer inputId: inputIds) {
  			streamGraph.addEdge(inputId,
  					sink.getId(),
  					0
  			);
  		}
  
  		if (sink.getStateKeySelector() != null) {
  			TypeSerializer<?> keySerializer = sink.getStateKeyType().createSerializer(env.getConfig());
  			streamGraph.setOneInputStateKey(sink.getId(), sink.getStateKeySelector(), keySerializer);
  		}
  
  		return Collections.emptyList();
  	}
docs: Transforms a {@code SourceTransformation}.