Skip to content

[FLINK-40067][tests] Fix race in RescaleTimelineITCase.testRescaleTerminatedByJobFinished #28633

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MartijnVisser wants to merge 1 commit into
base: master
Choose a base branch
from
+13 −3
Open

[FLINK-40067][tests] Fix race in RescaleTimelineITCase.testRescaleTerminatedByJobFinished #28633

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions ...test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,15 +269,24 @@ void testRescaleTerminatedBySucceeded() {}

@TestTemplate
void testRescaleTerminatedByJobFinished() throws Exception {
// This case only asserts on the recorded rescale history; skip the disabled-history
// parameter before the cluster rebuild below so it does not pay for an unused cluster.
assumeThat(enabledRescaleHistory(configuration)).isTrue();

// With the short shared cooldown the DefaultStateTransitionManager re-enters Idling on a
// wall-clock timer and terminates the in-progress rescale with
// NO_RESOURCES_OR_PARALLELISMS_CHANGE before the job finishes; goToFinished then finds it
// already terminated and its JOB_FINISHED stamp is ignored, so waiting cannot win.
// Widen the cooldown to keep the rescale in-progress until the job finishes.
rebuildClusterWithExecutingTimeouts(Duration.ofSeconds(60), Duration.ofSeconds(60));

final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);

miniCluster.submitJob(jobGraph).join();

waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID, PARALLELISM);

assumeThat(enabledRescaleHistory(configuration)).isTrue();

updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2);

// The upper bound (PARALLELISM * 2) exceeds the available slots, so this rescale is only
Expand All @@ -289,13 +298,14 @@ void testRescaleTerminatedByJobFinished() throws Exception {

OnceBlockingNoOpInvokable.unblock();

// Generous budget: on a loaded CI leg the unblock-to-finish window can itself exceed 10s.
waitUntilConditionWithTimeout(
() -> {
List<Rescale> rescaleHistory = getRescaleHistory(miniCluster, jobGraph);
return hasRescaleHistoryMetCondition(
rescaleHistory, 2, TerminatedReason.JOB_FINISHED);
},
10000);
60000);
}

@TestTemplate
Expand Down