[FLINK-1264] [compiler] Properly forward custom partitioners to the runtime
[FLINK-820] [compiler] Support for disconnected data flows