private DistributionPattern connectJobVertices(Channel channel, int inputNumber,
			final JobVertex sourceVertex, final TaskConfig sourceConfig,
			final JobVertex targetVertex, final TaskConfig targetConfig, boolean isBroadcast)
	throws CompilerException
	{
		// ------------ connect the vertices to the job graph --------------
		final DistributionPattern distributionPattern;

		switch (channel.getShipStrategy()) {
			case FORWARD:
				distributionPattern = DistributionPattern.POINTWISE;
				break;
			case PARTITION_RANDOM:
			case BROADCAST:
			case PARTITION_HASH:
			case PARTITION_CUSTOM:
			case PARTITION_RANGE:
			case PARTITION_FORCED_REBALANCE:
				distributionPattern = DistributionPattern.ALL_TO_ALL;
				break;
			default:
				throw new RuntimeException("Unknown runtime ship strategy: " + channel.getShipStrategy());
		}

		final ResultPartitionType resultType;

		switch (channel.getDataExchangeMode()) {

			case PIPELINED:
				resultType = ResultPartitionType.PIPELINED;
				break;

			case BATCH:
				// BLOCKING results are currently not supported in closed loop iterations
				//
				// See https://issues.apache.org/jira/browse/FLINK-1713 for details
				resultType = channel.getSource().isOnDynamicPath()
						? ResultPartitionType.PIPELINED
						: ResultPartitionType.BLOCKING;
				break;

			case PIPELINE_WITH_BATCH_FALLBACK:
				throw new UnsupportedOperationException("Data exchange mode " +
						channel.getDataExchangeMode() + " currently not supported.");

			default:
				throw new UnsupportedOperationException("Unknown data exchange mode.");

		}

		JobEdge edge = targetVertex.connectNewDataSetAsInput(sourceVertex, distributionPattern, resultType);

		// -------------- 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.");
			}
		}
		
		if (channel.getShipStrategy() == ShipStrategyType.PARTITION_CUSTOM) {
			if (channel.getPartitioner() != null) {
				sourceConfig.setOutputPartitioner(channel.getPartitioner(), outputIndex);
			} else {
				throw new CompilerException("The ship strategy was set to custom partitioning, but no partitioner was set.");
			}
		}
		
		// ---------------- configure the receiver -------------------
		if (isBroadcast) {
			targetConfig.addBroadcastInputToGroup(inputNumber);
		} else {
			targetConfig.addInputToGroup(inputNumber);
		}
		
		// ---------------- attach the additional infos to the job edge -------------------
		
		String shipStrategy = JsonMapper.getShipStrategyString(channel.getShipStrategy());
		if (channel.getShipStrategyKeys() != null && channel.getShipStrategyKeys().size() > 0) {
			shipStrategy += " on " + (channel.getShipStrategySortOrder() == null ?
					channel.getShipStrategyKeys().toString() :
					Utils.createOrdering(channel.getShipStrategyKeys(), channel.getShipStrategySortOrder()).toString());
		}
		
		String localStrategy;
		if (channel.getLocalStrategy() == null || channel.getLocalStrategy() == LocalStrategy.NONE) {
			localStrategy = null;
		}
		else {
			localStrategy = JsonMapper.getLocalStrategyString(channel.getLocalStrategy());
			if (localStrategy != null && channel.getLocalStrategyKeys() != null && channel.getLocalStrategyKeys().size() > 0) {
				localStrategy += " on " + (channel.getLocalStrategySortOrder() == null ?
						channel.getLocalStrategyKeys().toString() :
						Utils.createOrdering(channel.getLocalStrategyKeys(), channel.getLocalStrategySortOrder()).toString());
			}
		}
		
		String caching = channel.getTempMode() == TempMode.NONE ? null : channel.getTempMode().toString();

		edge.setShipStrategyName(shipStrategy);
		edge.setPreProcessingOperationName(localStrategy);
		edge.setOperatorLevelCachingDescription(caching);
		
		return distributionPattern;
	}
private DistributionPattern connectJobVertices ( Channel channel , int inputNumber , final JobVertex sourceVertex , final TaskConfig sourceConfig , final JobVertex targetVertex , final TaskConfig targetConfig , boolean isBroadcast ) throws CompilerException { final DistributionPattern distributionPattern ; switch ( channel . getShipStrategy ( ) ) { case FORWARD : distributionPattern = DistributionPattern . POINTWISE ; break ; case PARTITION_RANDOM : case BROADCAST : case PARTITION_HASH : case PARTITION_CUSTOM : case PARTITION_RANGE : case PARTITION_FORCED_REBALANCE : distributionPattern = DistributionPattern . ALL_TO_ALL ; break ; default : throw new RuntimeException ( "Unknown runtime ship strategy: " + channel . getShipStrategy ( ) ) ; } final ResultPartitionType resultType ; switch ( channel . getDataExchangeMode ( ) ) { case PIPELINED : resultType = ResultPartitionType . PIPELINED ; break ; case BATCH : resultType = channel . getSource ( ) . isOnDynamicPath ( ) ? ResultPartitionType . PIPELINED : ResultPartitionType . BLOCKING ; break ; case PIPELINE_WITH_BATCH_FALLBACK : throw new UnsupportedOperationException ( "Data exchange mode " + channel . getDataExchangeMode ( ) + " currently not supported." ) ; default : throw new UnsupportedOperationException ( "Unknown data exchange mode." ) ; } JobEdge edge = targetVertex . connectNewDataSetAsInput ( sourceVertex , distributionPattern , resultType ) ; 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 ( channel . getShipStrategy ( ) == ShipStrategyType . PARTITION_CUSTOM ) { if ( channel . getPartitioner ( ) != null ) { sourceConfig . setOutputPartitioner ( channel . getPartitioner ( ) , outputIndex ) ; } else { throw new CompilerException ( "The ship strategy was set to custom partitioning, but no partitioner was set." ) ; } } if ( isBroadcast ) { targetConfig . addBroadcastInputToGroup ( inputNumber ) ; } else { targetConfig . addInputToGroup ( inputNumber ) ; } String shipStrategy = JsonMapper . getShipStrategyString ( channel . getShipStrategy ( ) ) ; if ( channel . getShipStrategyKeys ( ) != null && channel . getShipStrategyKeys ( ) . size ( ) > 0 ) { shipStrategy += " on " + ( channel . getShipStrategySortOrder ( ) == null ? channel . getShipStrategyKeys ( ) . toString ( ) : Utils . createOrdering ( channel . getShipStrategyKeys ( ) , channel . getShipStrategySortOrder ( ) ) . toString ( ) ) ; } String localStrategy ; if ( channel . getLocalStrategy ( ) == null || channel . getLocalStrategy ( ) == LocalStrategy . NONE ) { localStrategy = null ; } else { localStrategy = JsonMapper . getLocalStrategyString ( channel . getLocalStrategy ( ) ) ; if ( localStrategy != null && channel . getLocalStrategyKeys ( ) != null && channel . getLocalStrategyKeys ( ) . size ( ) > 0 ) { localStrategy += " on " + ( channel . getLocalStrategySortOrder ( ) == null ? channel . getLocalStrategyKeys ( ) . toString ( ) : Utils . createOrdering ( channel . getLocalStrategyKeys ( ) , channel . getLocalStrategySortOrder ( ) ) . toString ( ) ) ; } } String caching = channel . getTempMode ( ) == TempMode . NONE ? null : channel . getTempMode ( ) . toString ( ) ; edge . setShipStrategyName ( shipStrategy ) ; edge . setPreProcessingOperationName ( localStrategy ) ; edge . setOperatorLevelCachingDescription ( caching ) ; return distributionPattern ; }
NOTE: The channel for global and local strategies are different if we connect a union. The global strategy
channel is then the channel into the union node, the local strategy channel the one from the union to the
actual target operator.

@throws CompilerException