Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2946#discussion_r103425262
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
---
@@ -642,31 +980,71 @@ public AllWindowedStream(DataStream<T> input,
* @return The data stream that is the result of applying the window
function to the window.
*/
public <R> SingleOutputStreamOperator<R> apply(AllWindowFunction<T, R,
W> function) {
+ String callLocation = Utils.getCallLocationName();
+ function = input.getExecutionEnvironment().clean(function);
TypeInformation<R> resultType =
TypeExtractor.getUnaryOperatorReturnType(
function, AllWindowFunction.class, true, true,
getInputType(), null, false);
-
- return apply(function, resultType);
+ return apply(new InternalIterableAllWindowFunction<>(function),
resultType, callLocation);
}
/**
* Applies the given window function to each window. The window
function is called for each
- * evaluation of the window for each key individually. The output of
the window function is
+ * evaluation of the window. The output of the window function is
* interpreted as a regular non-windowed stream.
*
* <p>
* Not that this function requires that all data in the windows is
buffered until the window
* is evaluated, as the function provides no means of incremental
aggregation.
*
* @param function The window function.
- * @param resultType Type information for the result type of the window
function
* @return The data stream that is the result of applying the window
function to the window.
*/
public <R> SingleOutputStreamOperator<R> apply(AllWindowFunction<T, R,
W> function, TypeInformation<R> resultType) {
+ String callLocation = Utils.getCallLocationName();
+ function = input.getExecutionEnvironment().clean(function);
+ return apply(new InternalIterableAllWindowFunction<>(function),
resultType, callLocation);
+ }
- //clean the closure
+ /**
+ * Applies the given window function to each window. The window
function is called for each
+ * evaluation of the window. The output of the window function is
+ * interpreted as a regular non-windowed stream.
+ *
+ * <p>
+ * Not that this function requires that all data in the windows is
buffered until the window
+ * is evaluated, as the function provides no means of incremental
aggregation.
+ *
+ * @param function The process window function.
+ * @return The data stream that is the result of applying the window
function to the window.
+ */
+ public <R> SingleOutputStreamOperator<R>
process(ProcessAllWindowFunction<T, R, W> function) {
--- End diff --
This and the following should be `@PublicEvolving`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
