guoweiM commented on a change in pull request #26:
URL: https://github.com/apache/flink-ml/pull/26#discussion_r748974927
##########
File path:
flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
##########
@@ -201,14 +207,23 @@ public void onEpochWatermarkIncrement(int epochWatermark)
throws IOException {
latestEpochWatermark = epochWatermark;
// Destroys all the operators with round < epoch watermark. Notes
that
- // the onEpochWatermarkIncrement must be from 0 and increment by 1
each time.
- if (wrappedOperators.containsKey(epochWatermark)) {
- try {
- closeStreamOperator(wrappedOperators.get(epochWatermark),
epochWatermark);
- } catch (Exception exception) {
- ExceptionUtils.rethrow(exception);
+ // the onEpochWatermarkIncrement must be from 0 and increment by 1
each time, except
+ // for the last round.
+ try {
+ if (epochWatermark < Integer.MAX_VALUE) {
+ S wrappedOperator =
wrappedOperators.remove(epochWatermark);
+ if (wrappedOperator != null) {
+ closeStreamOperator(wrappedOperator, epochWatermark,
epochWatermark);
+ }
+ } else {
Review comment:
Do you think that we could check the whether the `epochWatermark` < 0?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
