//----------------- connect source and target with edge ------------------------------
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);
	}
targetConfig.addInputToGroup(inputNumber);
}
//----------------- connect source and target with edge ------------------------------
FlinkEdge edge;
ShipStrategyType shipStrategy = channel.getShipStrategy();
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);
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);