sunhaibotb commented on a change in pull request #8742: [FLINK-11879] Add
validators for the uses of InputSelectable, BoundedOneInput and
BoundedMultiInput
URL: https://github.com/apache/flink/pull/8742#discussion_r301374518
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -661,18 +664,31 @@ private void configureCheckpointing() {
"exactly-once or at-least-once.");
}
- // --- configure the master-side checkpoint hooks ---
+ // --- configure the master-side checkpoint hooks and check
the types of operators---
final ArrayList<MasterTriggerRestoreHook.Factory> hooks = new
ArrayList<>();
for (StreamNode node : streamGraph.getStreamNodes()) {
- if (node.getOperatorFactory() instanceof
UdfStreamOperatorFactory) {
- Function f = ((UdfStreamOperatorFactory)
node.getOperatorFactory()).getUserFunction();
+ StreamOperatorFactory operatorFactory =
node.getOperatorFactory();
+ if (operatorFactory instanceof
UdfStreamOperatorFactory) {
+ Function f = ((UdfStreamOperatorFactory)
operatorFactory).getUserFunction();
if (f instanceof WithMasterCheckpointHook) {
hooks.add(new
FunctionMasterCheckpointHookFactory((WithMasterCheckpointHook<?>) f));
}
}
+
+ if (cfg.isCheckpointingEnabled() && operatorFactory !=
null) {
Review comment:
The following validation that you are talking about allows to enable the
forced checkpointing for the iterative streaming, which means that no exception
is thrown in that case. In this PR, we need to add the check to cover this
case, otherwise `IterateITCase#testWithCheckPointing` will fail.
```
if (isIterative() && checkpointConfig.isCheckpointingEnabled() &&
!checkpointConfig.isForceCheckpointing())
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
