Hi. I tried to use CoGroupedStreams w/ disableAutoGeneratedUIDs.
CoGroupedStreams creates two map operators without the ability to set uids
on them. These appear as "Map" in my operator graph. I noticed that the
CoGroupedStreams.apply function has two map calls without setting uids. If
I try to run with disableAutoGeneratedUIDs, I get the following error
"java.lang.IllegalStateException: Auto generated UIDs have been disabled
but no UID or hash has been assigned to operator Map".
How can I fix this? Extend the base CoGroupedStreams class?
```
public <T> DataStream<T> apply(CoGroupFunction<T1, T2, T> function,
TypeInformation<T> resultType) {
function =
(CoGroupFunction)this.input1.getExecutionEnvironment().clean(function);
CoGroupedStreams.UnionTypeInfo<T1, T2> unionType = new
CoGroupedStreams.UnionTypeInfo(this.input1.getType(),
this.input2.getType());
CoGroupedStreams.UnionKeySelector<T1, T2, KEY> unionKeySelector = new
CoGroupedStreams.UnionKeySelector(this.keySelector1, this.keySelector2);
DataStream<CoGroupedStreams.TaggedUnion<T1, T2>> taggedInput1 =
this.input1.map(new
CoGroupedStreams.Input1Tagger()).setParallelism(this.input1.getParallelism()).returns(unionType);
DataStream<CoGroupedStreams.TaggedUnion<T1, T2>> taggedInput2 =
this.input2.map(new
CoGroupedStreams.Input2Tagger()).setParallelism(this.input2.getParallelism()).returns(unionType);
DataStream<CoGroupedStreams.TaggedUnion<T1, T2>> unionStream =
taggedInput1.union(new DataStream[]{taggedInput2});
this.windowedStream = (new KeyedStream(unionStream, unionKeySelector,
this.keyType)).window(this.windowAssigner);
if (this.trigger != null) {
this.windowedStream.trigger(this.trigger);
}
if (this.evictor != null) {
this.windowedStream.evictor(this.evictor);
}
if (this.allowedLateness != null) {
this.windowedStream.allowedLateness(this.allowedLateness);
}
return this.windowedStream.apply(new
CoGroupedStreams.CoGroupWindowFunction(function), resultType);
}
```
