Ported to flink 0.6