[
https://issues.apache.org/jira/browse/FLINK-39387?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser updated FLINK-39387:
-----------------------------------
Fix Version/s: 2.2.2
> Fix flaky scheduler benchmark tests caused by thread assertion failure in
> async TDD creation
> --------------------------------------------------------------------------------------------
>
> Key: FLINK-39387
> URL: https://issues.apache.org/jira/browse/FLINK-39387
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Coordination
> Reporter: Liu
> Assignee: Liu
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.3.0, 2.2.2
>
>
> h2. Summary
> Fix flaky scheduler benchmark tests caused by asynchronous TDD creation
> h3. Problem
> After
> [FLINK-38114|https://issues.apache.org/jira/browse/FLINK-38114(https://issues.apache.org/jira/browse/FLINK-38114)]
> introduced asynchronous TDD (TaskDeploymentDescriptor) creation in
> {{{}Execution.deploy(){}}}, all 21 scheduler benchmark tests (e.g.,
> {{{}PartitionReleaseInBatchJobBenchmarkTest{}}}) become flaky with the
> following error:
> {code:java|title=Error Message}
> IllegalStateException: Cannot leave terminal state CANCELED to transition to
> SCHEDULED.
> {code}
> h3. Root Cause
> {{FLINK-38114}} changed {{Execution.deploy()}} to use an asynchronous flow:
> {code:java}
> CompletableFuture.supplyAsync(initOffloadedTaskRestoreRef(...), executor)
> // step 1: runs on scheduledExecutorService
> .thenComposeAsync(tryGetTaskDeploymentDescriptorForSlot(slot),
> jobMasterMainThreadExecutor) // step 2: dispatched to mainThreadExecutor
> {code}
> {{SchedulerBenchmarkUtils.createAndInitScheduler()}} uses
> {{ComponentMainThreadExecutorServiceAdapter.forMainThread()}} as the
> {{{}mainThreadExecutor{}}}. Internally, it wraps a
> {{DirectScheduledExecutorService}} whose {{execute()}} method runs the
> command *on the calling thread* and includes an assertion:
> {code:java}
> assert MainThreadValidatorUtil.isRunningInExpectedThread(main);
> {code}
> When step 1 completes on the {{scheduledExecutorService}} thread,
> {{thenComposeAsync}} dispatches the step 2 callback to
> {{{}jobMasterMainThreadExecutor.execute(callback){}}}. Since
> {{DirectScheduledExecutorService.execute()}} runs inline on the calling
> thread (which is the {{scheduledExecutorService}} thread, *not* the test main
> thread), the assertion fails with an {{{}AssertionError{}}}.
> This triggers the following chain:
> # {{AssertionError}} -> {{markFailed()}} -> {{processFail()}}
> # {{notifySchedulerNgAboutInternalTaskFailure()}} -> scheduler failover
> # Scheduler cancels all executions in the affected region (including sink
> executions)
> # Subsequent {{transitionState(SCHEDULED)}} on sink executions fails because
> they are already in {{CANCELED}} state.
> h3. Fix
> Following the same approach as
> [FLINK-38970|https://issues.apache.org/jira/browse/FLINK-38970(https://issues.apache.org/jira/browse/FLINK-38970)]
> (which fixed the identical issue in {{{}AdaptiveBatchSchedulerTest{}}}),
> replace {{ComponentMainThreadExecutorServiceAdapter.forMainThread()}} in
> {{SchedulerBenchmarkUtils}} with a {{SynchronousComponentMainThreadExecutor}}
> that:
> * Uses {{DirectScheduledExecutorService}} to execute tasks synchronously on
> the calling thread.
> * Overrides {{assertRunningInMainThread()}} as a no-op, skipping strict
> thread identity checks.
> * Avoids {{AssertionError}} when {{CompletableFuture}} callbacks are
> dispatched from background threads.
> h3. Affected Tests
> All 21 scheduler benchmark tests that use
> {{{}SchedulerBenchmarkUtils.createAndInitScheduler(){}}}:
> * {{PartitionReleaseInBatchJobBenchmarkTest}}
> * {{DeployingDownstreamTasksInBatchJobBenchmarkTest}}
> * {{SchedulingDownstreamTasksInBatchJobBenchmarkTest}}
> * ... and others under {{org.apache.flink.runtime.scheduler.benchmark}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
