Use Flink's redistribute for reshuffle.

Python needs to publish the URN over the FnAPI which is pretty easy, but Flink also needs to ensure that the composite structure does not get fused. Unlike with GBK, we can't assume all runners implement this as a primitive.