[FLINK-4016] FoldApplyWindowFunction is not properly initialized

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Duplicate
• Affects Version/s: 1.0.3
• Fix Version/s: 1.1.0
• Component/s: API / DataStream
• Labels: easyfix pull-request-available
• Flags: Important
• Assignee: Unassigned
• Reporter: RWenden
• Created: 03/Jun/16 14:50
• Updated: 14/Oct/19 11:21
• Resolved: 06/Jun/16 07:44

Description

FoldApplyWindowFunction's outputtype is not set.

We're using constructions like (excerpt):
.keyBy(0)
.countWindow(10, 5)
.fold(...)
Running this stream gives an runtime exception in FoldApplyWindowFunction:
"No initial value was serialized for the fold window function. Probably the setOutputType method was not called."

This can be easily fixed in WindowedStream.java by (around line# 449):
FoldApplyWindowFunction foldApplyWindowFunction = new FoldApplyWindowFunction<>(initialValue, foldFunction, function);
foldApplyWindowFunction.setOutputType(resultType, input.getExecutionConfig());

operator = new EvictingWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
keySel,
input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
stateDesc,
new InternalIterableWindowFunction<>(foldApplyWindowFunction),
trigger,
evictor);

Issue Links

• duplicates: FLINK-3977 Subclasses of InternalWindowFunction must support OutputTypeConfigurable
• links to: GitHub Pull Request #2070
• links to: GitHub Pull Request #2070