node_never_finishes (running_dataflow.rs:1069) checks a node's
top-level inputs map to decide if it's a source. That's wrong for a
runtime (operators:) node its inputs live under
operators[].config.inputs, so the top level is always empty. Every
operator node with real inputs gets misclassified as a source.
Impact is dataflow-wide, not per-node: select_finish_stragglers bails
out entirely (no escalation for anyone) the instant it sees one
never_finishes node. So one operator node permanently disarms the
watchdog for the whole graph if any other node later wedges, nothing
catches it.
Already found and knowingly deferred in #2957:
// NOTE: this misclassifies a runtime (`operators:`) node as a
// source ... Pre-existing and left alone here: correcting it
// would newly arm the straggler watchdog for operator dataflows,
// which is a behavior change well outside #2920.
Same PR fixed the identical mismatch elsewhere in this file
(is_drained/has_data_input) by reading registered inputs instead of
the descriptor. This spot never got the same fix.
Fix
let is_source = !self.has_data_input(node_id);
node_never_finishes(running_dataflow.rs:1069) checks a node'stop-level
inputsmap to decide if it's a source. That's wrong for aruntime (
operators:) node its inputs live underoperators[].config.inputs, so the top level is always empty. Everyoperator node with real inputs gets misclassified as a source.
Impact is dataflow-wide, not per-node:
select_finish_stragglersbailsout entirely (no escalation for anyone) the instant it sees one
never_finishesnode. So one operator node permanently disarms thewatchdog for the whole graph if any other node later wedges, nothing
catches it.
Already found and knowingly deferred in #2957:
Same PR fixed the identical mismatch elsewhere in this file
(
is_drained/has_data_input) by reading registered inputs instead ofthe descriptor. This spot never got the same fix.
Fix