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 00000000000000..06a9a109b5f60b --- /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 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 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 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 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 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 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 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 rescales, int expectedSize, TerminatedReason terminatedReasonOfLatest) { + return rescales.size() == expectedSize + && terminatedReasonOfLatest == rescales.get(0).getTerminatedReason(); + } + + private static List 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 e5920f282cbcff..486ff851477055 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 getClusterConfigs() { @@ -91,42 +68,10 @@ static Collection getClusterConfigs() { 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 @@ void testRecordRescaleForRecoverableFailover() throws Exception { @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 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 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 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 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 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 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 @@ void testRecordInProgressRescale() throws Exception { } // 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 @@ private static boolean enabledRescaleHistory(Configuration configuration) { 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 @@ private void assertVertexParallelismRescaleNotNullBesidesPreRelatedFields( assertThat(vpr.getSlotSharingGroupId()).isNotNull(); assertThat(vpr.getSlotSharingGroupName()).isNotNull(); } - - private void waitUntilConditionWithTimeout( - SupplierWithException 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 rescales, int expectedSize, TerminatedReason terminatedReasonOfLatest) { - return rescales.size() == expectedSize - && terminatedReasonOfLatest == rescales.get(0).getTerminatedReason(); - } - - private static List 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 00000000000000..0a94648250f3ab --- /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()); + } +}