// For forward edges, create as many tasks in upstream operator as in source operator
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);
	}
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());
}
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");
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);
}