code: private <IN1, IN2, OUT> Collection<Integer> transformTwoInputTransform(TwoInputTransformation<IN1, IN2, OUT> transform) {
  
  		Collection<Integer> inputIds1 = transform(transform.getInput1());
  		Collection<Integer> inputIds2 = transform(transform.getInput2());
  
  		// the recursive call might have already transformed this
  		if (alreadyTransformed.containsKey(transform)) {
  			return alreadyTransformed.get(transform);
  		}
  
  		List<Integer> allInputIds = new ArrayList<>();
  		allInputIds.addAll(inputIds1);
  		allInputIds.addAll(inputIds2);
  
  		String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), allInputIds);
  
  		streamGraph.addCoOperator(
  				transform.getId(),
  				slotSharingGroup,
  				transform.getCoLocationGroupKey(),
  				transform.getOperator(),
  				transform.getInputType1(),
  				transform.getInputType2(),
  				transform.getOutputType(),
  				transform.getName());
  
  		if (transform.getStateKeySelector1() != null || transform.getStateKeySelector2() != null) {
  			TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
  			streamGraph.setTwoInputStateKey(transform.getId(), transform.getStateKeySelector1(), transform.getStateKeySelector2(), keySerializer);
  		}
  
  		streamGraph.setParallelism(transform.getId(), transform.getParallelism());
  		streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());
  
  		for (Integer inputId: inputIds1) {
  			streamGraph.addEdge(inputId,
  					transform.getId(),
  					1
  			);
  		}
  
  		for (Integer inputId: inputIds2) {
  			streamGraph.addEdge(inputId,
  					transform.getId(),
  					2
  			);
  		}
  
  		return Collections.singleton(transform.getId());
  	}
docs: Transforms a {@code TwoInputTransformation}.
  
  <p>This recursively transforms the inputs, creates a new {@code StreamNode} in the graph and
  wired the inputs to this new node.