[FLINK-1264] [compiler] Properly forward custom partitioners to the r…

…untime

flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java

@@ -1072,19 +1072,21 @@ private DistributionPattern connectJobVertices(Channel channel, int inputNumber,
 		if (channel.getShipStrategy() == ShipStrategyType.PARTITION_RANGE) {
 
 			final DataDistribution dataDistribution = channel.getDataDistribution();
-			if(dataDistribution != null) {
+			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
 			}
 		}
-//		if (targetContract instanceof GenericDataSink) {
-//			final DataDistribution distri = ((GenericDataSink) targetContract).getDataDistribution();
-//			if (distri != null) {
-//				configForOutputShipStrategy.setOutputDataDistribution(distri);
-//			}
-//		}
+
+		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) {

flink-tests/src/test/java/org/apache/flink/test/misc/CustomPartitioningITCase.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.misc;
+
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.junit.Assert;
+
+@SuppressWarnings("serial")
+public class CustomPartitioningITCase extends JavaProgramTestBase {
+
+	@Override
+	protected void testProgram() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		if (!isCollectionExecution()) {
+			Assert.assertTrue(env.getDegreeOfParallelism() > 1);
+		}
+
+		env.generateSequence(1, 1000)
+			.partitionCustom(new AllZeroPartitioner(), new IdKeySelector<Long>())
+			.map(new FailExceptInPartitionZeroMapper())
+			.output(new DiscardingOutputFormat<Long>());
+
+		env.execute();
+	}
+
+	public static class FailExceptInPartitionZeroMapper extends RichMapFunction<Long, Long> {
+
+		@Override
+		public Long map(Long value) throws Exception {
+			if (getRuntimeContext().getIndexOfThisSubtask() == 0) {
+				return value;
+			} else {
+				throw new Exception("Received data in a partition other than partition 0");
+			}
+		}
+	}
+
+	public static class AllZeroPartitioner implements Partitioner<Long> {
+		@Override
+		public int partition(Long key, int numPartitions) {
+			return 0;
+		}
+	}
+
+	public static class IdKeySelector<T> implements KeySelector<T, T> {
+		@Override
+		public T getKey(T value) {
+			return value;
+		}
+	}
+}