You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2019/07/03 11:08:18 UTC
[GitHub] [flink] 1u0 commented on a change in pull request #8742:
[FLINK-11879] Add validators for the uses of InputSelectable,
BoundedOneInput and BoundedMultiInput
1u0 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_r299899202
##########
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:
I'd move this check (for all nodes) into a separate function and invoke the new function earlier (in `StreamingJobGraphGenerator.createJobGraph()` call or even before `StreamingJobGraphGenerator` is used).
`StreamGraph.getJobGraph()` already has a validation check before creating a `JobGraph`. Imo, it's a better place to carry such validation for now (excluding the case for `InputSelectable` and network credit-based flow control).
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services