[streaming] CoGroupedWindowReduceInvokable refactor