[FLINK-1110] Implement collection-based execution for coGroup