You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2025/02/18 01:32:21 UTC
(flink) 01/03: [hotfix][datastream] Add throwing signature to ProcessFunction methods
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6d53ce84f54c7e6cd84c79af6267761e66b54ecd
Author: Xu Huang <hu...@gmail.com>
AuthorDate: Tue Feb 11 00:33:09 2025 +0800
[hotfix][datastream] Add throwing signature to ProcessFunction methods
---
.../function/OneInputEventTimeStreamProcessFunction.java | 3 ++-
.../TwoInputBroadcastEventTimeStreamProcessFunction.java | 3 ++-
...woInputNonBroadcastEventTimeStreamProcessFunction.java | 3 ++-
.../function/TwoOutputEventTimeStreamProcessFunction.java | 3 ++-
.../api/function/OneInputStreamProcessFunction.java | 7 ++++---
.../function/TwoInputBroadcastStreamProcessFunction.java | 12 +++++++-----
.../TwoInputNonBroadcastStreamProcessFunction.java | 12 +++++++-----
.../api/function/TwoOutputStreamProcessFunction.java | 8 +++++---
.../EventTimeWrappedOneInputStreamProcessFunction.java | 10 ++++++----
...TimeWrappedTwoInputBroadcastStreamProcessFunction.java | 15 +++++++++------
...eWrappedTwoInputNonBroadcastStreamProcessFunction.java | 15 +++++++++------
.../EventTimeWrappedTwoOutputStreamProcessFunction.java | 11 +++++++----
12 files changed, 62 insertions(+), 40 deletions(-)
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/OneInputEventTimeStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/OneInputEventTimeStreamProcessFunction.java
index 86da2a867d0..6c499d1e392 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/OneInputEventTimeStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/OneInputEventTimeStreamProcessFunction.java
@@ -43,5 +43,6 @@ public interface OneInputEventTimeStreamProcessFunction<IN, OUT>
* Invoked when an event-time timer fires. Note that it is only used in {@link
* KeyedPartitionStream}.
*/
- default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputBroadcastEventTimeStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputBroadcastEventTimeStreamProcessFunction.java
index efa62f8a56f..abc8424e3b0 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputBroadcastEventTimeStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputBroadcastEventTimeStreamProcessFunction.java
@@ -43,5 +43,6 @@ public interface TwoInputBroadcastEventTimeStreamProcessFunction<IN1, IN2, OUT>
* Invoked when an event-time timer fires. Note that it is only used in {@link
* KeyedPartitionStream}.
*/
- default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputNonBroadcastEventTimeStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputNonBroadcastEventTimeStreamProcessFunction.java
index 1502825f781..2e7447f1990 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputNonBroadcastEventTimeStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoInputNonBroadcastEventTimeStreamProcessFunction.java
@@ -43,5 +43,6 @@ public interface TwoInputNonBroadcastEventTimeStreamProcessFunction<IN1, IN2, OU
* Invoked when an event-time timer fires. Note that it is only used in {@link
* KeyedPartitionStream}.
*/
- default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoOutputEventTimeStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoOutputEventTimeStreamProcessFunction.java
index 56e33db636f..950417755bb 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoOutputEventTimeStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/extension/eventtime/function/TwoOutputEventTimeStreamProcessFunction.java
@@ -50,5 +50,6 @@ public interface TwoOutputEventTimeStreamProcessFunction<IN, OUT1, OUT2>
long timestamp,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputPartitionedContext<OUT1, OUT2> ctx) {}
+ TwoOutputPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java
index ad1a20454bf..c81eb6799e5 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java
@@ -56,7 +56,7 @@ public interface OneInputStreamProcessFunction<IN, OUT> extends ProcessFunction
*
* @param ctx the context in which this function is executed.
*/
- default void endInput(NonPartitionedContext<OUT> ctx) {}
+ default void endInput(NonPartitionedContext<OUT> ctx) throws Exception {}
/**
* Callback for processing timer.
@@ -66,11 +66,12 @@ public interface OneInputStreamProcessFunction<IN, OUT> extends ProcessFunction
* @param ctx runtime context in which this function is executed.
*/
default void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {}
/** Callback function when receive watermark. */
default WatermarkHandlingResult onWatermark(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java
index fc583dacacc..3d7ae3db82d 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java
@@ -70,7 +70,7 @@ public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends P
*
* @param ctx the context in which this function is executed.
*/
- default void endNonBroadcastInput(NonPartitionedContext<OUT> ctx) {}
+ default void endNonBroadcastInput(NonPartitionedContext<OUT> ctx) throws Exception {}
/**
* This is a life-cycle method indicates that this function will no longer receive any data from
@@ -78,7 +78,7 @@ public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends P
*
* @param ctx the context in which this function is executed.
*/
- default void endBroadcastInput(NonPartitionedContext<OUT> ctx) {}
+ default void endBroadcastInput(NonPartitionedContext<OUT> ctx) throws Exception {}
/**
* Callback for processing timer.
@@ -88,7 +88,7 @@ public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends P
* @param ctx runtime context in which this function is executed.
*/
default void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {}
/**
* Callback function when receive the watermark from broadcast input.
@@ -98,7 +98,8 @@ public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends P
* @param ctx runtime context in which this function is executed.
*/
default WatermarkHandlingResult onWatermarkFromBroadcastInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
@@ -110,7 +111,8 @@ public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends P
* @param ctx runtime context in which this function is executed.
*/
default WatermarkHandlingResult onWatermarkFromNonBroadcastInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java
index 356cc97e2fb..274fb661960 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java
@@ -66,7 +66,7 @@ public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extend
*
* @param ctx the context in which this function is executed.
*/
- default void endFirstInput(NonPartitionedContext<OUT> ctx) {}
+ default void endFirstInput(NonPartitionedContext<OUT> ctx) throws Exception {}
/**
* This is a life-cycle method indicates that this function will no longer receive any data from
@@ -74,7 +74,7 @@ public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extend
*
* @param ctx the context in which this function is executed.
*/
- default void endSecondInput(NonPartitionedContext<OUT> ctx) {}
+ default void endSecondInput(NonPartitionedContext<OUT> ctx) throws Exception {}
/**
* Callback for processing timer.
@@ -84,7 +84,7 @@ public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extend
* @param ctx runtime context in which this function is executed.
*/
default void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {}
/**
* Callback function when receive the watermark from the first input.
@@ -94,7 +94,8 @@ public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extend
* @param ctx runtime context in which this function is executed.
*/
default WatermarkHandlingResult onWatermarkFromFirstInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
@@ -106,7 +107,8 @@ public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extend
* @param ctx runtime context in which this function is executed.
*/
default WatermarkHandlingResult onWatermarkFromSecondInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
}
diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java
index 1136ebc5053..079fb1487b8 100644
--- a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java
+++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java
@@ -61,7 +61,7 @@ public interface TwoOutputStreamProcessFunction<IN, OUT1, OUT2> extends ProcessF
*
* @param ctx the context in which this function is executed.
*/
- default void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) {}
+ default void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) throws Exception {}
/**
* Callback for processing timer.
@@ -75,7 +75,8 @@ public interface TwoOutputStreamProcessFunction<IN, OUT1, OUT2> extends ProcessF
long timestamp,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputPartitionedContext<OUT1, OUT2> ctx) {}
+ TwoOutputPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {}
/**
* Callback function when receive the watermark from the input.
@@ -89,7 +90,8 @@ public interface TwoOutputStreamProcessFunction<IN, OUT1, OUT2> extends ProcessF
Watermark watermark,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) {
+ TwoOutputNonPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {
return WatermarkHandlingResult.PEEK;
}
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedOneInputStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedOneInputStreamProcessFunction.java
index b1867af7a39..083795bea28 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedOneInputStreamProcessFunction.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedOneInputStreamProcessFunction.java
@@ -83,19 +83,20 @@ public class EventTimeWrappedOneInputStreamProcessFunction<IN, OUT>
}
@Override
- public void endInput(NonPartitionedContext<OUT> ctx) {
+ public void endInput(NonPartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.endInput(ctx);
}
@Override
public void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.onProcessingTimer(timestamp, output, ctx);
}
@Override
public WatermarkHandlingResult onWatermark(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -118,7 +119,8 @@ public class EventTimeWrappedOneInputStreamProcessFunction<IN, OUT>
}
}
- public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext ctx) {
+ public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {
wrappedUserFunction.onEventTimer(timestamp, output, ctx);
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputBroadcastStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputBroadcastStreamProcessFunction.java
index db3c256482b..dfd2d2a8932 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputBroadcastStreamProcessFunction.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputBroadcastStreamProcessFunction.java
@@ -91,24 +91,25 @@ public class EventTimeWrappedTwoInputBroadcastStreamProcessFunction<IN1, IN2, OU
}
@Override
- public void endBroadcastInput(NonPartitionedContext<OUT> ctx) {
+ public void endBroadcastInput(NonPartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.endBroadcastInput(ctx);
}
@Override
- public void endNonBroadcastInput(NonPartitionedContext<OUT> ctx) {
+ public void endNonBroadcastInput(NonPartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.endNonBroadcastInput(ctx);
}
@Override
public void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.onProcessingTimer(timestamp, output, ctx);
}
@Override
public WatermarkHandlingResult onWatermarkFromBroadcastInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -133,7 +134,8 @@ public class EventTimeWrappedTwoInputBroadcastStreamProcessFunction<IN1, IN2, OU
@Override
public WatermarkHandlingResult onWatermarkFromNonBroadcastInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -155,7 +157,8 @@ public class EventTimeWrappedTwoInputBroadcastStreamProcessFunction<IN1, IN2, OU
}
}
- public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {
+ public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {
wrappedUserFunction.onEventTimer(timestamp, output, ctx);
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction.java
index a8dc809a90e..dd9aa63fc3e 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction.java
@@ -91,24 +91,25 @@ public class EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction<IN1, IN2,
}
@Override
- public void endFirstInput(NonPartitionedContext<OUT> ctx) {
+ public void endFirstInput(NonPartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.endFirstInput(ctx);
}
@Override
- public void endSecondInput(NonPartitionedContext<OUT> ctx) {
+ public void endSecondInput(NonPartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.endSecondInput(ctx);
}
@Override
public void onProcessingTimer(
- long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {
+ long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception {
wrappedUserFunction.onProcessingTimer(timestamp, output, ctx);
}
@Override
public WatermarkHandlingResult onWatermarkFromFirstInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -132,7 +133,8 @@ public class EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction<IN1, IN2,
@Override
public WatermarkHandlingResult onWatermarkFromSecondInput(
- Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx) {
+ Watermark watermark, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -154,7 +156,8 @@ public class EventTimeWrappedTwoInputNonBroadcastStreamProcessFunction<IN1, IN2,
}
}
- public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {
+ public void onEventTime(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {
wrappedUserFunction.onEventTimer(timestamp, output, ctx);
}
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoOutputStreamProcessFunction.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoOutputStreamProcessFunction.java
index 5a272f31588..8e6e0f17203 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoOutputStreamProcessFunction.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/eventtime/functions/EventTimeWrappedTwoOutputStreamProcessFunction.java
@@ -87,7 +87,7 @@ public class EventTimeWrappedTwoOutputStreamProcessFunction<IN, OUT1, OUT2>
}
@Override
- public void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) {
+ public void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) throws Exception {
wrappedUserFunction.endInput(ctx);
}
@@ -96,7 +96,8 @@ public class EventTimeWrappedTwoOutputStreamProcessFunction<IN, OUT1, OUT2>
long timestamp,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputPartitionedContext<OUT1, OUT2> ctx) {
+ TwoOutputPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {
wrappedUserFunction.onProcessingTimer(timestamp, output1, output2, ctx);
}
@@ -105,7 +106,8 @@ public class EventTimeWrappedTwoOutputStreamProcessFunction<IN, OUT1, OUT2>
Watermark watermark,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) {
+ TwoOutputNonPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {
if (EventTimeExtensionImpl.isEventTimeExtensionWatermark(watermark)) {
// If the watermark is from the event time extension, process it and call {@code
// userFunction#onEventTimeWatermark} when the event time is updated; otherwise, forward
@@ -131,7 +133,8 @@ public class EventTimeWrappedTwoOutputStreamProcessFunction<IN, OUT1, OUT2>
long timestamp,
Collector<OUT1> output1,
Collector<OUT2> output2,
- TwoOutputPartitionedContext<OUT1, OUT2> ctx) {
+ TwoOutputPartitionedContext<OUT1, OUT2> ctx)
+ throws Exception {
wrappedUserFunction.onEventTimer(timestamp, output1, output2, ctx);
}