private void connectJobVertices(Channel channel, int inputNumber,
							final FlinkVertex sourceVertex, final TezTaskConfig sourceConfig,
							final FlinkVertex targetVertex, final TezTaskConfig targetConfig, boolean isBroadcast)
			throws CompilerException {
		// -------------- configure the source task's ship strategy strategies in task config --------------
		final int outputIndex = sourceConfig.getNumOutputs();
		sourceConfig.addOutputShipStrategy(channel.getShipStrategy());
		if (outputIndex == 0) {
			sourceConfig.setOutputSerializer(channel.getSerializer());
		}
		if (channel.getShipStrategyComparator() != null) {
			sourceConfig.setOutputComparator(channel.getShipStrategyComparator(), outputIndex);
		}
		if (channel.getShipStrategy() == ShipStrategyType.PARTITION_RANGE) {
			final DataDistribution dataDistribution = channel.getDataDistribution();
			if(dataDistribution != null) {
				sourceConfig.setOutputDataDistribution(dataDistribution, outputIndex);
			} else {
				throw new RuntimeException("Range partitioning requires data distribution");
				// TODO: inject code and configuration for automatic histogram generation
			}
		}
		// ---------------- configure the receiver -------------------
		if (isBroadcast) {
			targetConfig.addBroadcastInputToGroup(inputNumber);
		} else {
			targetConfig.addInputToGroup(inputNumber);
		}
		//----------------- connect source and target with edge ------------------------------
		FlinkEdge edge;
		ShipStrategyType shipStrategy = channel.getShipStrategy();
		TypeSerializer<?> serializer = channel.getSerializer().getSerializer();
		if ((shipStrategy == ShipStrategyType.FORWARD) || (shipStrategy == ShipStrategyType.NONE)) {
			edge = new FlinkForwardEdge(sourceVertex, targetVertex, serializer);
			// For forward edges, create as many tasks in upstream operator as in source operator
			targetVertex.setParallelism(sourceVertex.getParallelism());
		}
		else if (shipStrategy == ShipStrategyType.BROADCAST) {
			edge = new FlinkBroadcastEdge(sourceVertex, targetVertex, serializer);
		}
		else if (shipStrategy == ShipStrategyType.PARTITION_HASH) {
			edge = new FlinkPartitionEdge(sourceVertex, targetVertex, serializer);
		}
		else {
			throw new CompilerException("Ship strategy between nodes " + sourceVertex.getVertex().getName() + " and " + targetVertex.getVertex().getName() + " currently not supported");
		}
		// Tez-specific bookkeeping
		// TODO: This probably will not work for vertices with multiple outputs
		sourceVertex.addNumberOfSubTasksInOutput(targetVertex.getParallelism(), outputIndex);
		targetVertex.addInput(sourceVertex, inputNumber);
		edges.add(edge);
	}
private void connectJobVertices(Channel channel, int inputNumber,
							final FlinkVertex sourceVertex, final TezTaskConfig sourceConfig,
							final FlinkVertex targetVertex, final TezTaskConfig targetConfig, boolean isBroadcast)
			throws CompilerException {
	
		final int outputIndex = sourceConfig.getNumOutputs();
		sourceConfig.addOutputShipStrategy(channel.getShipStrategy());
		if (outputIndex == 0) {
			sourceConfig.setOutputSerializer(channel.getSerializer());
		}
		if (channel.getShipStrategyComparator() != null) {
			sourceConfig.setOutputComparator(channel.getShipStrategyComparator(), outputIndex);
		}
		if (channel.getShipStrategy() == ShipStrategyType.PARTITION_RANGE) {
			final DataDistribution dataDistribution = channel.getDataDistribution();
			if(dataDistribution != null) {
				sourceConfig.setOutputDataDistribution(dataDistribution, outputIndex);
			} else {
				throw new RuntimeException("Range partitioning requires data distribution");
			
			}
		}
	
		if (isBroadcast) {
			targetConfig.addBroadcastInputToGroup(inputNumber);
		} else {
			targetConfig.addInputToGroup(inputNumber);
		}
	
		FlinkEdge edge;
		ShipStrategyType shipStrategy = channel.getShipStrategy();
		TypeSerializer<?> serializer = channel.getSerializer().getSerializer();
		if ((shipStrategy == ShipStrategyType.FORWARD) || (shipStrategy == ShipStrategyType.NONE)) {
			edge = new FlinkForwardEdge(sourceVertex, targetVertex, serializer);
		
			targetVertex.setParallelism(sourceVertex.getParallelism());
		}
		else if (shipStrategy == ShipStrategyType.BROADCAST) {
			edge = new FlinkBroadcastEdge(sourceVertex, targetVertex, serializer);
		}
		else if (shipStrategy == ShipStrategyType.PARTITION_HASH) {
			edge = new FlinkPartitionEdge(sourceVertex, targetVertex, serializer);
		}
		else {
			throw new CompilerException("Ship strategy between nodes " + sourceVertex.getVertex().getName() + " and " + targetVertex.getVertex().getName() + " currently not supported");
		}
	
	
		sourceVertex.addNumberOfSubTasksInOutput(targetVertex.getParallelism(), outputIndex);
		targetVertex.addInput(sourceVertex, inputNumber);
		edges.add(edge);
	}
private void connectjobvertices(channel channel, int inputnumber, final flinkvertex sourcevertex, final teztaskconfig sourceconfig, final flinkvertex targetvertex, final teztaskconfig targetconfig, boolean isbroadcast) throws compilerexception { final int outputindex = sourceconfig.getnumoutputs(); sourceconfig.addoutputshipstrategy(channel.getshipstrategy()); if (outputindex == 0) { sourceconfig.setoutputserializer(channel.getserializer()); } if (channel.getshipstrategycomparator() != null) { sourceconfig.setoutputcomparator(channel.getshipstrategycomparator(), outputindex); } if (channel.getshipstrategy() == shipstrategytype.partition_range) { final datadistribution datadistribution = channel.getdatadistribution(); if(datadistribution != null) { sourceconfig.setoutputdatadistribution(datadistribution, outputindex); } else { throw new runtimeexception("range partitioning requires data distribution"); } } if (isbroadcast) { targetconfig.addbroadcastinputtogroup(inputnumber); } else { targetconfig.addinputtogroup(inputnumber); } flinkedge edge; shipstrategytype shipstrategy = channel.getshipstrategy(); typeserializer<?> serializer = channel.getserializer().getserializer(); if ((shipstrategy == shipstrategytype.forward) || (shipstrategy == shipstrategytype.none)) { edge = new flinkforwardedge(sourcevertex, targetvertex, serializer); targetvertex.setparallelism(sourcevertex.getparallelism()); } else if (shipstrategy == shipstrategytype.broadcast) { edge = new flinkbroadcastedge(sourcevertex, targetvertex, serializer); } else if (shipstrategy == shipstrategytype.partition_hash) { edge = new flinkpartitionedge(sourcevertex, targetvertex, serializer); } else { throw new compilerexception("ship strategy between nodes " + sourcevertex.getvertex().getname() + " and " + targetvertex.getvertex().getname() + " currently not supported"); } sourcevertex.addnumberofsubtasksinoutput(targetvertex.getparallelism(), outputindex); targetvertex.addinput(sourcevertex, inputnumber); edges.add(edge); }