This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7063101f454 [FLINK-40150][tests] De-parameterize
rescale-history-enabled RescaleTimelineITCase tests
7063101f454 is described below
commit 7063101f45454eddfa0fdb2778dce7c0473046d6
Author: Martijn Visser <[email protected]>
AuthorDate: Wed Jul 15 13:39:34 2026 +0200
[FLINK-40150][tests] De-parameterize rescale-history-enabled
RescaleTimelineITCase tests
The terminated-reason tests only assert when rescale history is enabled, yet
ran as @TestTemplate: on the disabled parameter each paid for a MiniCluster
startup before assumeThat aborted, and three rebuilt the cluster on top.
Move
them to a non-parameterized RescaleTimelineHistoryEnabledITCase that starts
one
cluster per test with the configuration it needs. Shared setup and helpers
move
to RescaleTimelineITCaseBase; RescaleTimelineITCase keeps the parameterized
cases.
Generated-by: Claude Code (Claude Opus 4.8)
---
.../RescaleTimelineHistoryEnabledITCase.java | 273 ++++++++++++++++
.../adaptive/timeline/RescaleTimelineITCase.java | 363 +--------------------
.../timeline/RescaleTimelineITCaseBase.java | 156 +++++++++
3 files changed, 435 insertions(+), 357 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineHistoryEnabledITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineHistoryEnabledITCase.java
new file mode 100644
index 00000000000..06a9a109b5f
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineHistoryEnabledITCase.java
@@ -0,0 +1,273 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive.timeline;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.WebOptions;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
+import org.apache.flink.runtime.testtasks.OnceBlockingNoOpInvokable;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for rescale terminated reasons and terminal state, only meaningful
when rescale history is
+ * enabled. Each case starts its own cluster with the configuration it needs.
+ */
+class RescaleTimelineHistoryEnabledITCase extends RescaleTimelineITCaseBase {
+
+ // Matches the default poll interval of the
CommonTestUtils#waitUntilCondition it replaced.
+ private static final long RETRY_INTERVAL_MILLIS = 100L;
+
+ @BeforeEach
+ void setUp() {
+ OnceBlockingNoOpInvokable.reset();
+ }
+
+ @Test
+ void testRescaleTerminatedByJobFinished() throws Exception {
+ // 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.
+ startCluster(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);
+
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
+
+ // The upper bound (PARALLELISM * 2) exceeds the available slots, so
this rescale is only
+ // recorded in the history without changing the parallelism. Wait
until it is recorded
+ // before unblocking, otherwise on a slow machine the task can finish
first and the size-2
+ // condition below times out.
+ waitUntilConditionWithTimeout(
+ () -> getRescaleHistory(miniCluster, jobGraph).size() == 2,
10000);
+
+ 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);
+ },
+ 60000);
+ }
+
+ @Test
+ void testRescaleTerminatedByJobCancelled() throws Exception {
+ startCluster(createEnabledConfiguration());
+
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+ final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
+
+ miniCluster.submitJob(jobGraph).join();
+
+ waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
+
+ miniCluster.cancelJob(jobGraph.getJobID());
+
+ waitUntilConditionWithTimeout(
+ () -> {
+ List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
+ return hasRescaleHistoryMetCondition(
+ rescaleHistory, 2, TerminatedReason.JOB_CANCELED);
+ },
+ 10000);
+ }
+
+ @Test
+ void testRescaleTerminatedByJobFailed() throws Exception {
+ startCluster(createEnabledConfiguration());
+
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+ final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
+
+ miniCluster.submitJob(jobGraph).join();
+
+ waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+
+ miniCluster.terminateTaskManager(1);
+
+ waitForVertexParallelismReachedAndJobRunning(
+ jobGraph, JOB_VERTEX_ID, NUMBER_SLOTS_PER_TASK_MANAGER);
+
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
+
+ miniCluster.terminateTaskManager(0);
+
+ waitUntilConditionWithTimeout(
+ () -> {
+ List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
+ return hasRescaleHistoryMetCondition(
+ rescaleHistory, 3, TerminatedReason.JOB_FAILED);
+ },
+ 10000);
+ }
+
+ @Test
+ void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws
Exception {
+ // NO_RESOURCES_OR_PARALLELISMS_CHANGE is only stamped when the update
RPC is processed
+ // during Cooldown and the manager re-enters Idling. Unlike the sibling
+ // testRescaleTerminatedByResourceRequirementsUpdated, this case must
wait out the whole
+ // cooldown, so it cannot reuse 60s: 10s outlasts the RPC yet stays
within the 60s budget.
+ startCluster(Duration.ofSeconds(10), Duration.ofMillis(50));
+
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+ final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
+ miniCluster.submitJob(jobGraph).join();
+ waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
+
+ waitUntilConditionWithTimeout(
+ () -> {
+ List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
+ return hasRescaleHistoryMetCondition(
+ rescaleHistory,
+ 2,
+
TerminatedReason.NO_RESOURCES_OR_PARALLELISMS_CHANGE);
+ },
+ 60000);
+ }
+
+ @Test
+ void testRescaleTerminatedByResourcesNotEnoughException() throws Exception
{
+ startCluster(createEnabledConfiguration());
+
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+ final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
+ miniCluster.submitJob(jobGraph).join();
+ waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+
+ miniCluster.terminateTaskManager(1);
+ miniCluster.terminateTaskManager(0);
+
+ waitUntilConditionWithTimeout(
+ () -> {
+ List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
+ return hasRescaleHistoryMetCondition(
+ rescaleHistory, 2,
TerminatedReason.EXCEPTION_OCCURRED);
+ },
+ 100000);
+ }
+
+ @Test
+ void testRescaleTerminatedByResourceRequirementsUpdated() throws Exception
{
+ // The second update only terminates the first rescale with
RESOURCE_REQUIREMENTS_UPDATED
+ // while that rescale is still in-progress; once it is terminated,
updateRescale is a no-op.
+ // With the short shared cooldown the manager re-enters Idling and
terminates it on a
+ // wall-clock timer first, so waiting cannot win the race. Widening
cooldown and
+ // stabilization to 60s keeps the rescale in-progress far longer than
the RPC round trip.
+ startCluster(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);
+
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
+ updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
3);
+
+ waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+
+ final ExecutionGraphInfo executionGraphInfo =
+ miniCluster.getExecutionGraphInfo(jobGraph.getJobID()).join();
+ assertThat(executionGraphInfo.getRescalesStatsSnapshot()).isNotNull();
+ List<Rescale> rescaleHistory =
+
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory();
+ assertThat(rescaleHistory).hasSize(3);
+ Rescale rescale = rescaleHistory.get(1);
+ assertTerminalRelatedFields(rescale,
TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
+ }
+
+ // Private methods.
+ private static Configuration createEnabledConfiguration() {
+ final Configuration configuration = createConfiguration();
+
configuration.set(WebOptions.MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE, 3);
+ return configuration;
+ }
+
+ /**
+ * Starts the fixture cluster with rescale history enabled plus the given
executing-phase
+ * cooldown and resource-stabilization timeouts, kept wide enough to
observe a rescale's
+ * terminated reason before the default short cooldown races past it.
+ */
+ private void startCluster(Duration cooldown, Duration
resourceStabilizationTimeout)
+ throws Exception {
+ final Configuration configuration = createEnabledConfiguration();
+
configuration.set(JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING,
cooldown);
+ configuration.set(
+
JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT,
+ resourceStabilizationTimeout);
+ startCluster(configuration);
+ }
+
+ private void waitUntilConditionWithTimeout(
+ SupplierWithException<Boolean, Exception> condition, long
timeoutMillis)
+ throws Exception {
+ // Poll synchronously via waitUtil. The previous
CompletableFuture#runAsync + get(timeout)
+ // wrapper hangs on JDK 11: on a ForkJoinPool worker (JUnit's
executor) timedGet
+ // help-executes the unbounded poll loop inline on the waiting thread,
so the timeout
+ // never fires.
+ org.apache.flink.core.testutils.CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ return condition.get();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ },
+ Duration.ofMillis(timeoutMillis),
+ Duration.ofMillis(RETRY_INTERVAL_MILLIS),
+ "Condition was not met within " + timeoutMillis + " ms.");
+ }
+
+ private boolean hasRescaleHistoryMetCondition(
+ List<Rescale> rescales, int expectedSize, TerminatedReason
terminatedReasonOfLatest) {
+ return rescales.size() == expectedSize
+ && terminatedReasonOfLatest ==
rescales.get(0).getTerminatedReason();
+ }
+
+ private static List<Rescale> getRescaleHistory(MiniCluster miniCluster,
JobGraph jobGraph)
+ throws InterruptedException, ExecutionException {
+ ExecutionGraphInfo executionGraphInfo =
+ miniCluster.getExecutionGraphInfo(jobGraph.getJobID()).get();
+ return
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory();
+ }
+}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java
index e5920f282cb..486ff851477 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java
@@ -18,68 +18,45 @@
package org.apache.flink.runtime.scheduler.adaptive.timeline;
-import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.WebOptions;
-import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
-import org.apache.flink.runtime.jobgraph.JobResourceRequirements;
-import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.minicluster.MiniCluster;
import
org.apache.flink.runtime.rest.messages.job.rescales.JobRescaleDetails.VertexParallelismRescaleInfo;
import org.apache.flink.runtime.rest.messages.job.rescales.SchedulerStateSpan;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import org.apache.flink.runtime.testtasks.OnceBlockingNoOpInvokable;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.runtime.testutils.MiniClusterResource;
-import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.util.function.RunnableWithException;
-import org.apache.flink.util.function.SupplierWithException;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
-import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assumptions.assumeThat;
/**
- * Test for recording rescale history by {@link DefaultRescaleTimeline} or
{@link
- * RescaleTimeline.NoOpRescaleTimeline}.
+ * Tests for recording rescale history by {@link DefaultRescaleTimeline} or
{@link
+ * RescaleTimeline.NoOpRescaleTimeline}. Each case runs once with rescale
history disabled and once
+ * with it enabled, asserting the parameter-appropriate outcome. Cases that
are only meaningful with
+ * rescale history enabled live in {@link RescaleTimelineHistoryEnabledITCase}.
*/
@ExtendWith(ParameterizedTestExtension.class)
-class RescaleTimelineITCase {
-
- private static final int NUMBER_SLOTS_PER_TASK_MANAGER = 2;
- private static final int NUMBER_TASK_MANAGERS = 2;
- private static final int PARALLELISM = NUMBER_SLOTS_PER_TASK_MANAGER *
NUMBER_TASK_MANAGERS;
- private static final JobVertexID JOB_VERTEX_ID = new JobVertexID();
- // Matches the default poll interval of the
CommonTestUtils#waitUntilCondition it replaced.
- private static final long RETRY_INTERVAL_MILLIS = 100L;
+class RescaleTimelineITCase extends RescaleTimelineITCaseBase {
@Parameter private Configuration configuration;
- private MiniClusterResource miniClusterResource;
@Parameters
static Collection<Object[]> getClusterConfigs() {
@@ -91,42 +68,10 @@ class RescaleTimelineITCase {
new Object[] {confEnabledRescaleHistory});
}
- private static Configuration createConfiguration() {
- final Configuration configuration = new Configuration();
- configuration.set(JobManagerOptions.SCHEDULER,
JobManagerOptions.SchedulerType.Adaptive);
- configuration.set(
-
JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT,
- Duration.ofMillis(50));
- configuration.set(
- JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING,
- // Use the 0.1 seconds to trigger the long-time non-terminal
rescale event after a
- // rescaling.
- Duration.ofMillis(100));
- configuration.set(
-
JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT,
- Duration.ofMillis(50));
- configuration.set(
- JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_WAIT_TIMEOUT,
- Duration.ofSeconds(2));
- return configuration;
- }
-
@BeforeEach
void setUp() throws Exception {
OnceBlockingNoOpInvokable.reset();
- this.miniClusterResource =
- new MiniClusterResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(configuration)
-
.setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
- .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
- .build());
- miniClusterResource.before();
- }
-
- @AfterEach
- void tearDown() {
- miniClusterResource.after();
+ startCluster(configuration);
}
// Tests for rescale trigger causes.
@@ -267,186 +212,6 @@ class RescaleTimelineITCase {
@TestTemplate
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);
-
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
-
- // The upper bound (PARALLELISM * 2) exceeds the available slots, so
this rescale is only
- // recorded in the history without changing the parallelism. Wait
until it is recorded
- // before unblocking, otherwise on a slow machine the task can finish
first and the size-2
- // condition below times out.
- waitUntilConditionWithTimeout(
- () -> getRescaleHistory(miniCluster, jobGraph).size() == 2,
10000);
-
- 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);
- },
- 60000);
- }
-
- @TestTemplate
- void testRescaleTerminatedByJobCancelled() throws Exception {
- final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
- final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
-
- miniCluster.submitJob(jobGraph).join();
-
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
-
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
-
- miniCluster.cancelJob(jobGraph.getJobID());
-
- assumeThat(enabledRescaleHistory(configuration)).isTrue();
- waitUntilConditionWithTimeout(
- () -> {
- List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
- return hasRescaleHistoryMetCondition(
- rescaleHistory, 2, TerminatedReason.JOB_CANCELED);
- },
- 10000);
- }
-
- @TestTemplate
- void testRescaleTerminatedByJobFailed() throws Exception {
- final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
- final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
-
- miniCluster.submitJob(jobGraph).join();
-
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
-
- miniCluster.terminateTaskManager(1);
-
- waitForVertexParallelismReachedAndJobRunning(
- jobGraph, JOB_VERTEX_ID, NUMBER_SLOTS_PER_TASK_MANAGER);
-
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
-
- miniCluster.terminateTaskManager(0);
-
- assumeThat(enabledRescaleHistory(configuration)).isTrue();
- waitUntilConditionWithTimeout(
- () -> {
- List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
- return hasRescaleHistoryMetCondition(
- rescaleHistory, 3, TerminatedReason.JOB_FAILED);
- },
- 10000);
- }
-
- @TestTemplate
- void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() 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();
-
- // NO_RESOURCES_OR_PARALLELISMS_CHANGE is only stamped when the update
RPC is processed
- // during Cooldown and the manager re-enters Idling. Unlike the sibling
- // testRescaleTerminatedByResourceRequirementsUpdated, this case must
wait out the whole
- // cooldown, so it cannot reuse 60s: 10s outlasts the RPC yet stays
within the 60s budget.
- rebuildClusterWithExecutingTimeouts(Duration.ofSeconds(10),
Duration.ofMillis(50));
-
- final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
- final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
- miniCluster.submitJob(jobGraph).join();
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
-
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
-
- waitUntilConditionWithTimeout(
- () -> {
- List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
- return hasRescaleHistoryMetCondition(
- rescaleHistory,
- 2,
-
TerminatedReason.NO_RESOURCES_OR_PARALLELISMS_CHANGE);
- },
- 60000);
- }
-
- @TestTemplate
- void testRescaleTerminatedByResourcesNotEnoughException() throws Exception
{
- final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
- final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM);
- miniCluster.submitJob(jobGraph).join();
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
-
- miniCluster.terminateTaskManager(1);
- miniCluster.terminateTaskManager(0);
-
- assumeThat(enabledRescaleHistory(configuration)).isTrue();
- waitUntilConditionWithTimeout(
- () -> {
- List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
- return hasRescaleHistoryMetCondition(
- rescaleHistory, 2,
TerminatedReason.EXCEPTION_OCCURRED);
- },
- 100000);
- }
-
- @TestTemplate
- void testRescaleTerminatedByResourceRequirementsUpdated() 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();
-
- // The second update only terminates the first rescale with
RESOURCE_REQUIREMENTS_UPDATED
- // while that rescale is still in-progress; once it is terminated,
updateRescale is a no-op.
- // With the short shared cooldown the manager re-enters Idling and
terminates it on a
- // wall-clock timer first, so waiting cannot win the race. Widening
cooldown and
- // stabilization to 60s keeps the rescale in-progress far longer than
the RPC round trip.
- 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);
-
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
2);
- updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM *
3);
-
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
-
- final ExecutionGraphInfo executionGraphInfo =
- miniCluster.getExecutionGraphInfo(jobGraph.getJobID()).join();
- runAdaptedParameterizedAssertion(
- executionGraphInfo,
- () -> {
-
assertThat(executionGraphInfo.getRescalesStatsSnapshot()).isNotNull();
- List<Rescale> rescaleHistory =
-
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory();
- assertThat(rescaleHistory).hasSize(3);
- Rescale rescale = rescaleHistory.get(1);
- assertTerminalRelatedFields(
- rescale,
TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
- });
- }
-
/**
* The test case is disabled after processing by
* https://lists.apache.org/thread/hh7w2p6lnmbo1q6d9ngkttdyrw4lp74h. Merge
the current
@@ -624,18 +389,6 @@ class RescaleTimelineITCase {
}
// Private methods.
- private JobGraph createBlockingJobGraph(int parallelism) {
- final JobVertex blockingOperator = new JobVertex("Blocking operator",
JOB_VERTEX_ID);
- SlotSharingGroup sharingGroup = new SlotSharingGroup();
- sharingGroup.setSlotSharingGroupName("slot-sharing-group-A");
- blockingOperator.setSlotSharingGroup(sharingGroup);
- blockingOperator.setInvokableClass(OnceBlockingNoOpInvokable.class);
- blockingOperator.setParallelism(parallelism);
- final JobGraph jobGraph =
JobGraphTestUtils.streamingJobGraph(blockingOperator);
- RestartStrategyUtils.configureFixedDelayRestartStrategy(jobGraph, 1,
0L);
- return jobGraph;
- }
-
private void runAdaptedParameterizedAssertion(
ExecutionGraphInfo executionGraphInfo,
RunnableWithException assertionForEnabledRescaleHistory)
@@ -652,15 +405,6 @@ class RescaleTimelineITCase {
return
configuration.get(WebOptions.MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE) > 0;
}
- private static void assertTerminalRelatedFields(
- Rescale rescale, TerminatedReason expectedTerminatedReason) {
- TerminatedReason terminatedReason = rescale.getTerminatedReason();
- assertThat(terminatedReason).isEqualTo(expectedTerminatedReason);
- assertThat(terminatedReason.getTerminalState())
- .isEqualTo(rescale.getTerminalState())
- .isEqualTo(expectedTerminatedReason.getTerminalState());
- }
-
private static void assertTriggerCause(Rescale rescale, TriggerCause
expectedTriggerCause) {
assertThat(rescale).isNotNull();
assertThat(rescale.getTriggerCause()).isEqualTo(expectedTriggerCause);
@@ -687,99 +431,4 @@ class RescaleTimelineITCase {
assertThat(vpr.getSlotSharingGroupId()).isNotNull();
assertThat(vpr.getSlotSharingGroupName()).isNotNull();
}
-
- private void waitUntilConditionWithTimeout(
- SupplierWithException<Boolean, Exception> condition, long
timeoutMillis)
- throws Exception {
- // Poll synchronously via waitUtil. The previous
CompletableFuture#runAsync + get(timeout)
- // wrapper hangs on JDK 11: on a ForkJoinPool worker (JUnit's
executor) timedGet
- // help-executes the unbounded poll loop inline on the waiting thread,
so the timeout
- // never fires.
- org.apache.flink.core.testutils.CommonTestUtils.waitUtil(
- () -> {
- try {
- return condition.get();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- },
- Duration.ofMillis(timeoutMillis),
- Duration.ofMillis(RETRY_INTERVAL_MILLIS),
- "Condition was not met within " + timeoutMillis + " ms.");
- }
-
- private void waitForVertexParallelismReachedAndJobRunning(
- JobGraph jobGraph, JobVertexID jobVertexId, int targetParallelism)
throws Exception {
- CommonTestUtils.waitUntilCondition(
- () -> {
- final ArchivedExecutionGraph archivedExecutionGraph =
- miniClusterResource
- .getMiniCluster()
-
.getArchivedExecutionGraph(jobGraph.getJobID())
- .get();
- final AccessExecutionJobVertex executionJobVertex =
-
archivedExecutionGraph.getAllVertices().get(jobVertexId);
- if (executionJobVertex == null) {
- // parallelism was not yet determined
- return false;
- }
- return executionJobVertex.getParallelism() ==
targetParallelism;
- });
- CommonTestUtils.waitUntilCondition(
- () ->
-
miniClusterResource.getMiniCluster().getJobStatus(jobGraph.getJobID()).get()
- == JobStatus.RUNNING);
- }
-
- /**
- * Rebuilds the shared fixture cluster in place with the given
executing-phase cooldown and
- * resource-stabilization timeouts. Rebuilding in place rather than
starting a second cluster
- * keeps a single cluster running so the {@link AfterEach} teardown still
applies.
- */
- private void rebuildClusterWithExecutingTimeouts(
- Duration cooldown, Duration resourceStabilizationTimeout) throws
Exception {
- miniClusterResource.after();
- final Configuration testConfiguration = new
Configuration(configuration);
- testConfiguration.set(
-
JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING, cooldown);
- testConfiguration.set(
-
JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT,
- resourceStabilizationTimeout);
- miniClusterResource =
- new MiniClusterResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(testConfiguration)
-
.setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
- .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
- .build());
- miniClusterResource.before();
- }
-
- private void updateJobResourceRequirements(
- MiniCluster miniCluster, JobGraph jobGraph, int lowerBound, int
upperBound)
- throws ExecutionException, InterruptedException {
- miniCluster
- .updateJobResourceRequirements(
- jobGraph.getJobID(),
- JobResourceRequirements.newBuilder()
- .setParallelismForJobVertex(
-
jobGraph.getVertices().iterator().next().getID(),
- lowerBound,
- upperBound)
- .build())
- .get();
- }
-
- private boolean hasRescaleHistoryMetCondition(
- List<Rescale> rescales, int expectedSize, TerminatedReason
terminatedReasonOfLatest) {
- return rescales.size() == expectedSize
- && terminatedReasonOfLatest ==
rescales.get(0).getTerminatedReason();
- }
-
- private static List<Rescale> getRescaleHistory(MiniCluster miniCluster,
JobGraph jobGraph)
- throws InterruptedException, ExecutionException {
- ExecutionGraphInfo executionGraphInfo =
- miniCluster.getExecutionGraphInfo(jobGraph.getJobID()).get();
- return
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory();
- }
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCaseBase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCaseBase.java
new file mode 100644
index 00000000000..0a94648250f
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCaseBase.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive.timeline;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
+import org.apache.flink.runtime.jobgraph.JobResourceRequirements;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.testtasks.OnceBlockingNoOpInvokable;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.util.RestartStrategyUtils;
+
+import org.junit.jupiter.api.AfterEach;
+
+import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Shared fixture and helpers for the {@link RescaleTimeline} integration
tests, reused by {@link
+ * RescaleTimelineITCase} and {@link RescaleTimelineHistoryEnabledITCase}.
+ */
+abstract class RescaleTimelineITCaseBase {
+
+ protected static final int NUMBER_SLOTS_PER_TASK_MANAGER = 2;
+ protected static final int NUMBER_TASK_MANAGERS = 2;
+ protected static final int PARALLELISM = NUMBER_SLOTS_PER_TASK_MANAGER *
NUMBER_TASK_MANAGERS;
+ protected static final JobVertexID JOB_VERTEX_ID = new JobVertexID();
+
+ protected MiniClusterResource miniClusterResource;
+
+ protected static Configuration createConfiguration() {
+ final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.SCHEDULER,
JobManagerOptions.SchedulerType.Adaptive);
+ configuration.set(
+
JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT,
+ Duration.ofMillis(50));
+ configuration.set(
+ JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING,
+ // Use the 0.1 seconds to trigger the long-time non-terminal
rescale event after a
+ // rescaling.
+ Duration.ofMillis(100));
+ configuration.set(
+
JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT,
+ Duration.ofMillis(50));
+ configuration.set(
+ JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_WAIT_TIMEOUT,
+ Duration.ofSeconds(2));
+ return configuration;
+ }
+
+ /** Builds and starts the fixture cluster with the given configuration. */
+ protected void startCluster(Configuration configuration) throws Exception {
+ miniClusterResource =
+ new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(configuration)
+
.setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
+ .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
+ .build());
+ miniClusterResource.before();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (miniClusterResource != null) {
+ miniClusterResource.after();
+ }
+ }
+
+ protected JobGraph createBlockingJobGraph(int parallelism) {
+ final JobVertex blockingOperator = new JobVertex("Blocking operator",
JOB_VERTEX_ID);
+ SlotSharingGroup sharingGroup = new SlotSharingGroup();
+ sharingGroup.setSlotSharingGroupName("slot-sharing-group-A");
+ blockingOperator.setSlotSharingGroup(sharingGroup);
+ blockingOperator.setInvokableClass(OnceBlockingNoOpInvokable.class);
+ blockingOperator.setParallelism(parallelism);
+ final JobGraph jobGraph =
JobGraphTestUtils.streamingJobGraph(blockingOperator);
+ RestartStrategyUtils.configureFixedDelayRestartStrategy(jobGraph, 1,
0L);
+ return jobGraph;
+ }
+
+ protected void waitForVertexParallelismReachedAndJobRunning(
+ JobGraph jobGraph, JobVertexID jobVertexId, int targetParallelism)
throws Exception {
+ CommonTestUtils.waitUntilCondition(
+ () -> {
+ final ArchivedExecutionGraph archivedExecutionGraph =
+ miniClusterResource
+ .getMiniCluster()
+
.getArchivedExecutionGraph(jobGraph.getJobID())
+ .get();
+ final AccessExecutionJobVertex executionJobVertex =
+
archivedExecutionGraph.getAllVertices().get(jobVertexId);
+ if (executionJobVertex == null) {
+ // parallelism was not yet determined
+ return false;
+ }
+ return executionJobVertex.getParallelism() ==
targetParallelism;
+ });
+ CommonTestUtils.waitUntilCondition(
+ () ->
+
miniClusterResource.getMiniCluster().getJobStatus(jobGraph.getJobID()).get()
+ == JobStatus.RUNNING);
+ }
+
+ protected void updateJobResourceRequirements(
+ MiniCluster miniCluster, JobGraph jobGraph, int lowerBound, int
upperBound)
+ throws ExecutionException, InterruptedException {
+ miniCluster
+ .updateJobResourceRequirements(
+ jobGraph.getJobID(),
+ JobResourceRequirements.newBuilder()
+ .setParallelismForJobVertex(
+
jobGraph.getVertices().iterator().next().getID(),
+ lowerBound,
+ upperBound)
+ .build())
+ .get();
+ }
+
+ protected static void assertTerminalRelatedFields(
+ Rescale rescale, TerminatedReason expectedTerminatedReason) {
+ TerminatedReason terminatedReason = rescale.getTerminatedReason();
+ assertThat(terminatedReason).isEqualTo(expectedTerminatedReason);
+ assertThat(terminatedReason.getTerminalState())
+ .isEqualTo(rescale.getTerminalState())
+ .isEqualTo(expectedTerminatedReason.getTerminalState());
+ }
+}
