[
https://issues.apache.org/jira/browse/FLINK-40078?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099823#comment-18099823
]
Iman Rastkhadiv commented on FLINK-40078:
-----------------------------------------
Hi, I'd like to work on this one if nobody else has picked it up.
I went through the path you describe and it lines up:
OperatorEvent.isLossTolerant() defaults to false (OperatorEvent.java:32), and
SubtaskGatewayImpl only returns early when the failure is a
TaskNotRunningException and the event is loss-tolerant
(SubtaskGatewayImpl.java:117); otherwise it wraps the failure in the "An
OperatorEvent ... was lost" FlinkException and calls
subtaskAccess.triggerTaskFailover. WatermarkAlignmentEvent is currently the
only OperatorEvent overriding isLossTolerant() to true, so NoMoreSplitsEvent
takes the escalation path.
Between the two directions you listed, my instinct is that the second is the
one that actually fixes the class of problem: marking NoMoreSplitsEvent
loss-tolerant removes the symptom for that one event, but any other coordinator
event in flight when a task fails can still overwrite the original cause. That
said, it is the more invasive change and it touches failover consistency
semantics, so I would rather have a runtime/coordination owner's read on the
intended behaviour before I commit to an approach.
Could a committer assign this to me? I am happy to start with whichever
direction you prefer, with a test that reproduces the cause masking.
> Lost non-loss-tolerant OperatorEvent to an already-failing task triggers a
> failover that masks the original failure cause
> -------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40078
> URL: https://issues.apache.org/jira/browse/FLINK-40078
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Reporter: Martijn Visser
> Priority: Major
>
> When an OperatorCoordinator sends an event to a subtask that has just failed
> for an application reason, the delivery fails with
> {{TaskNotRunningException}} ({{Task.deliverOperatorEvent}}). Because the
> event is not loss-tolerant, {{SubtaskGatewayImpl.sendEvent}} escalates this
> into
> {code}
> org.apache.flink.util.FlinkException: An OperatorEvent from an
> OperatorCoordinator to a task was lost. Triggering task failover to ensure
> consistency. Event: '[NoMoreSplitEvent]', ...
> {code}
> and triggers a task failover. Under {{NoRestartBackoffTimeStrategy}} that
> lost-event failover can win the race to the scheduler and become the job's
> reported terminal failure cause, completely replacing the original
> application error in the cause chain (the coordinator is co-located with the
> JobMaster, so its failure path can reach the scheduler faster than the failed
> task's own failure RPC).
> The consequence is that a user debugging a failed job sees "An OperatorEvent
> ... was lost. Triggering task failover to ensure consistency" instead of
> their actual error. This is general: any coordinator event in flight at the
> moment a task fails for an application reason can mask that failure. It was
> observed concretely in FLINK-40077, where a {{NoMoreSplitsEvent}} in flight
> to a task that intentionally failed on the NOT NULL enforcer replaced the
> enforcer error as the reported cause (there it was worked around in the test
> scaffolding; this ticket is about the runtime behavior itself).
> The escalation mechanism already has an escape hatch: on
> {{TaskNotRunningException}}, {{SubtaskGatewayImpl}} returns without failover
> when {{evt.isLossTolerant()}} is true. {{NoMoreSplitsEvent}} is currently not
> marked loss-tolerant.
> Possible directions (needs input from runtime/coordination owners, as this
> touches checkpoint/failover consistency semantics):
> * Mark {{NoMoreSplitsEvent}} loss-tolerant, since a task that is FAILED or
> FINISHED has no further use for it.
> * More generally, suppress the redundant lost-event failover when the target
> task is already failing/failed for another reason, so the original cause is
> preserved.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
