@Test
public void testOutputTypeConfigurationWithTwoInputTransformation() throws Exception {

	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<Integer> source1 = env.fromElements(1, 10);
	DataStream<Integer> source2 = env.fromElements(2, 11);
	ConnectedStreams<Integer, Integer> connectedSource = source1.connect(source2);
	OutputTypeConfigurableOperationWithTwoInputs outputTypeConfigurableOperation = new OutputTypeConfigurableOperationWithTwoInputs();
	DataStream<Integer> result = connectedSource.transform(
			"Two input and output type configurable operation",
			BasicTypeInfo.INT_TYPE_INFO,
			outputTypeConfigurableOperation);
	result.addSink(new DiscardingSink<Integer>());
	env.getStreamGraph();
	assertEquals(BasicTypeInfo.INT_TYPE_INFO, outputTypeConfigurableOperation.getTypeInformation());
}