Rion Williams created FLINK-41505:
-------------------------------------
Summary: Loss-tolerant OperatorEvent that hits a finished task
fails the coordinator checkpoint
Key: FLINK-41505
URL: https://issues.apache.org/jira/browse/FLINK-41505
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 2.1.3, 2.2.1, 2.3.0
Reporter: Rion Williams
Fix For: 2.5.0
h3. Problem
FLINK-37670 made {{WatermarkAlignmentEvent}} loss tolerant
({{OperatorEvent#isLossTolerant}}) and has the {{SourceCoordinator}} keep
sending it after the enumerator has no more splits.
{{SubtaskGatewayImpl#sendEvent}} accordingly skips the task failover when such
an event fails with {{TaskNotRunningException}}.
However, the future of that event is still registered in the
{{IncompleteFuturesTracker}} and completes exceptionally. If a coordinator
checkpoint is waiting for pending events at that moment,
{{OperatorCoordinatorHolder#completeCheckpointOnceEventsAreDone}} fails it:
{noformat}
Coordinator checkpoint <id> for coordinator <op> is awaiting 1 pending events
...
org.apache.flink.util.FlinkException: Failing OperatorCoordinator checkpoint
because some OperatorEvents before this checkpoint barrier were not received by
the target tasks.
{noformat}
So an event loss that the gateway treats as harmless still fails the checkpoint.
h3. When it happens
Watermark alignment is enabled, a source subtask finishes while an alignment
announcement (every {{updateInterval}}) is in flight to it, and a coordinator
checkpoint is taken before that delivery fails. The window is when source
subtasks finish, which is also when coordinator snapshots tend to be taken:
* streaming with bounded or finishing sources (checkpoints after tasks
finished): a failed checkpoint, which counts towards the tolerable checkpoint
failures;
* batch job recovery: {{DefaultBatchJobRecoveryHandler#onExecutionFinished}}
snapshots the coordinators (checkpoint id -1) when an execution finishes, so
the snapshot future fails while the {{ExecutionVertexFinishedEvent}} is written.
We found it through a job that takes a single coordinator snapshot right after
the sources reach end of input and cannot retry it, so the race failed the
whole job in about 1 in 5 runs.
h3. Reproduction
Deterministic unit test against {{OperatorCoordinatorHolderTest}}. It fails on
master at the last assertion:
{code:java}
@Test
void testCheckpointSucceedsIfLossTolerantEventFindsTaskNotRunning() throws
Exception {
CompletableFuture<Acknowledge> eventSendingResult = new
CompletableFuture<>();
final EventReceivingTasks tasks =
EventReceivingTasks.createForRunningTasksWithRpcResult(eventSendingResult);
final OperatorCoordinatorHolder holder =
createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);
// Send one loss-tolerant event without finishing it.
CompletableFuture<Acknowledge> sendResult =
getCoordinator(holder).getSubtaskGateway(0).sendEvent(new
TestOperatorEvent(0, true));
// Trigger one checkpoint.
CompletableFuture<byte[]> checkpointResult = new CompletableFuture<>();
holder.checkpointCoordinator(1, checkpointResult);
getCoordinator(holder).getLastTriggeredCheckpoint().complete(new byte[0]);
// The target task finished before the event reached it.
eventSendingResult.completeExceptionally(
new FlinkException(new TaskNotRunningException("test")));
assertThat(sendResult).isCompletedExceptionally();
assertThat(checkpointResult).isCompleted(); // fails: completed
exceptionally
}
{code}
h3. Proposed fix
In {{SubtaskGatewayImpl#sendEvent}}, track a future for the checkpoint that
treats a loss-tolerant event failing with {{TaskNotRunningException}} as
delivered, i.e. the same condition that already skips the failover. The future
returned to the caller still completes exceptionally, so
{{SubtaskGatewayImplTest#optionalEventsIgnoreTaskNotRunning}} is unchanged. It
is a small change in {{SubtaskGatewayImpl}} plus the test above; I have a patch
ready.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
