You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "RocMarshal (via GitHub)" <gi...@apache.org> on 2026/02/05 16:00:23 UTC
[PR] [FLINK-38342][runtime] Support record rescale information. [flink]
RocMarshal opened a new pull request, #27539:
URL: https://github.com/apache/flink/pull/27539
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
[FLINK-38342][runtime] Support record rescale information.
## Brief change log
- [FLINK-38342][runtime] Support record rescale information.
- Add the test cases placeholders to prepare testing in the FLINK-38343 PR.
## Verifying this change
Will add the test cases placeholders to prepare testing in the FLINK-38343 PR.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r2872310223
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -31,8 +32,17 @@ public interface StateTransitionManager {
/**
* Is called if the environment changed in a way that a state transition could be considered.
+ *
+ * @param newResourceDriven Whether the onchange is driven by new available resources.
Review Comment:
JavaDoc typo/wording: `onchange` should be `onChange`, and the sentence can be clarified (e.g., “Whether the onChange is driven by newly available resources.”).
```suggestion
* @param newResourceDriven Whether the onChange is driven by newly available resources.
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java:
##########
@@ -2699,6 +2700,22 @@ public DummyState(StateWithoutExecutionGraph.Context context, JobStatus jobStatu
this.jobStatus = jobStatus;
}
+ @Override
+ public Durable getDurable() {
+ return null;
+ }
+
+ @Override
+ public void cancel() {}
+
+ @Override
+ public void suspend(Throwable cause) {}
+
+ @Override
+ public JobID getJobId() {
+ return null;
Review Comment:
These overrides in `DummyState` return `null` / no-op for methods that are already correctly implemented by `StateWithoutExecutionGraph` (e.g., `getDurable()` is used by `State#onLeave` and must not be null). This can cause NPEs and makes the test state behave unlike real states. Remove these overrides or delegate to `super`/`context` instead of returning null.
```suggestion
return super.getDurable();
}
@Override
public void cancel() {
super.cancel();
}
@Override
public void suspend(Throwable cause) {
super.suspend(cause);
}
@Override
public JobID getJobId() {
return super.getJobId();
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1101,6 +1130,27 @@ public void updateJobResourceRequirements(JobResourceRequirements jobResourceReq
}
}
+ private void recordRescaleForNewResourceRequirements() {
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.addSchedulerState(state)
+ .setTerminatedReason(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED)
+ .setEndTimestamp(Instant.now().toEpochMilli())
+ .log());
Review Comment:
In `recordRescaleForNewResourceRequirements()`, the first `updateRescale(...)` will emit WARN logs and be ignored if the timeline is idling/terminated. If this is expected, please guard the update with `!rescaleTimeline.isIdling()` to keep logs clean (similar to the `goToCanceling` guard).
```suggestion
if (!rescaleTimeline.isIdling()) {
rescaleTimeline.updateRescale(
rescale ->
rescale.addSchedulerState(state)
.setTerminatedReason(
TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED)
.setEndTimestamp(Instant.now().toEpochMilli())
.log());
}
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1349,6 +1452,12 @@ public CompletableFuture<String> goToStopWithSavepoint(
@Override
public void goToFinished(ArchivedExecutionGraph archivedExecutionGraph) {
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.addSchedulerState(state)
+ .setEndTimestamp(Instant.now().toEpochMilli())
+ .setTerminatedReason(TerminatedReason.JOB_FINISHED)
+ .log());
Review Comment:
`goToFinished(...)` updates the rescale timeline unconditionally. If there is no active rescale (idling/terminated), `DefaultRescaleTimeline.updateRescale()` logs WARNs and ignores the update; this can lead to noisy logs on every normal job completion. Consider guarding with `if (!rescaleTimeline.isIdling())` (or otherwise ensuring a rescale exists) before updating.
```suggestion
if (!rescaleTimeline.isIdling()) {
rescaleTimeline.updateRescale(
rescale ->
rescale.addSchedulerState(state)
.setEndTimestamp(Instant.now().toEpochMilli())
.setTerminatedReason(TerminatedReason.JOB_FINISHED)
.log());
}
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1304,13 +1365,55 @@ public void goToRestarting(
}
}
+ private void recordRescaleForJobRestarting(VertexParallelism restartWithParallelism) {
+ if (restartWithParallelism == null) {
+ // For the failover restarting.
+ if (!rescaleTimeline.isIdling()) {
+ // Process by https://lists.apache.org/thread/hh7w2p6lnmbo1q6d9ngkttdyrw4lp74h.
+ LOG.info(
+ "Merge the current non-terminated rescale and the new rescale triggered by recoverable failover into the current rescale.");
+ rescaleTimeline.updateRescale(Rescale::clearSchedulerStates);
+ } else if (rescaleTimeline.isIdling()) {
+ rescaleTimeline.newRescale(false);
+ }
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.setStartTimestamp(Instant.now().toEpochMilli())
+ .setTriggerCause(TriggerCause.RECOVERABLE_FAILOVER)
+ .setMinimalRequiredSlots(jobInformation)
+ .setPreRescaleSlotsAndParallelisms(
+ jobInformation,
+ rescaleTimeline.getLatestRescale(
+ TerminalState.COMPLETED))
+ .setDesiredVertexParallelism(jobInformation)
+ .setDesiredSlots(jobInformation)
+ .log());
+ } else {
+ // For the normal rescaling restarting.
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.setMinimalRequiredSlots(jobInformation)
+ .setDesiredVertexParallelism(jobInformation)
+ .setDesiredSlots(jobInformation)
+ .log());
+ }
+ }
+
@Override
public void goToFailing(
ExecutionGraph executionGraph,
ExecutionGraphHandler executionGraphHandler,
OperatorCoordinatorHandler operatorCoordinatorHandler,
Throwable failureCause,
List<ExceptionHistoryEntry> failureCollection) {
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.setEndTimestamp(Instant.now().toEpochMilli())
+ .addSchedulerState(state, failureCause)
+ .setTerminatedReason(TerminatedReason.JOB_FAILED)
+ .setStringifiedException(
+ ExceptionUtils.stringifyException(failureCause))
+ .log());
Review Comment:
`goToFailing(...)` updates the rescale timeline unconditionally. If there is no active rescale (idling/terminated), `DefaultRescaleTimeline.updateRescale()` logs WARNs and ignores the update; this can create noisy logs during normal job failures unrelated to rescaling. Consider guarding with `if (!rescaleTimeline.isIdling())` (like `goToCanceling`).
```suggestion
if (!rescaleTimeline.isIdling()) {
rescaleTimeline.updateRescale(
rescale ->
rescale.setEndTimestamp(Instant.now().toEpochMilli())
.addSchedulerState(state, failureCause)
.setTerminatedReason(TerminatedReason.JOB_FAILED)
.setStringifiedException(
ExceptionUtils.stringifyException(failureCause))
.log());
}
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1618,6 +1727,9 @@ <T extends State> T transitionToState(StateFactory<T> targetState) {
final JobStatus previousJobStatus = state.getJobStatus();
state.onLeave(targetState.getStateClass());
+
+ rescaleTimeline.updateRescale(rescale -> rescale.addSchedulerState(state));
Review Comment:
`transitionToState(...)` always calls `rescaleTimeline.updateRescale(...)` on state transitions. When there is no active rescale, `DefaultRescaleTimeline.updateRescale()` logs WARNs and ignores the update, potentially producing warnings on many (or all) state transitions. Consider guarding this call with `if (!rescaleTimeline.isIdling())` or making the timeline implementation tolerate idling updates without WARNs.
```suggestion
if (!rescaleTimeline.isIdling()) {
rescaleTimeline.updateRescale(rescale -> rescale.addSchedulerState(state));
}
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test for recording rescale history by {@link DefaultRescaleTimeline} or {@link
+ * RescaleTimeline.NoOpRescaleTimeline}.
+ */
+class RescaleTimelineITCase {
+ static final String DISABLED_DESCRIPTION =
+ "TODO: Blocked by FLINK-38343, the ITCases need the SchedulerNG#requstJob() to get the rescale history.";
Review Comment:
The disabled-test description contains a typo: `SchedulerNG#requstJob()` should be `SchedulerNG#requestJob()` (and consider fixing the sentence grammar while you are here).
```suggestion
"TODO: Blocked by FLINK-38343. These IT cases require SchedulerNG#requestJob() to get the rescale history.";
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java:
##########
@@ -511,6 +512,16 @@ public TestingStateTransitionManagerContext withSufficientResources() {
// StateTransitionManager.Context interface methods
// ///////////////////////////////////////////////
+ @Override
+ public State schedulerstate() {
+ throw new UnsupportedOperationException();
Review Comment:
`schedulerstate()` currently throws `UnsupportedOperationException`. Some new production paths call this method (e.g., rescale-recording logic in `DefaultStateTransitionManager.onChange(boolean)`), which can make these tests fail unexpectedly once `onChange(true)` is exercised. Prefer returning a stable stub `State` (or `null`) that won’t trigger rescale-recording branches, instead of throwing.
```suggestion
return null;
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java:
##########
@@ -292,14 +307,30 @@ static final class Idling extends Phase {
private Idling(Supplier<Temporal> clock, DefaultStateTransitionManager context) {
super(clock, context);
+ recordRescaleForNoResourcesOrNoParallelismChange();
}
@Override
- void onChange() {
+ void onChange(boolean newResourceDriven) {
if (hasSufficientResources()) {
+ recordRescaleForNewAvailableResources(context(), newResourceDriven);
context().progressToStabilizing(now());
}
}
+
+ private void recordRescaleForNoResourcesOrNoParallelismChange() {
+ if (context().transitionContext.schedulerstate() instanceof Executing) {
+ context()
+ .getRescaleTimeline()
+ .updateRescale(
+ rescale ->
+ rescale.setTerminatedReason(
+ TerminatedReason
+ .NO_RESOURCES_OR_PARALLELISMS_CHANGE)
+ .setEndTimestamp(Instant.now().toEpochMilli())
+ .addSchedulerState(context().schedulerState()));
+ }
Review Comment:
`recordRescaleForNoResourcesOrNoParallelismChange()` updates the rescale unconditionally once the scheduler state is `Executing`, but `RescaleTimeline.updateRescale()` logs WARNs when there is no active rescale (idling/terminated) and ignores the update. Consider guarding this with `!getRescaleTimeline().isIdling()` (or otherwise ensuring a rescale exists) to avoid noisy logs and no-op updates.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1304,13 +1365,55 @@ public void goToRestarting(
}
}
+ private void recordRescaleForJobRestarting(VertexParallelism restartWithParallelism) {
+ if (restartWithParallelism == null) {
+ // For the failover restarting.
+ if (!rescaleTimeline.isIdling()) {
+ // Process by https://lists.apache.org/thread/hh7w2p6lnmbo1q6d9ngkttdyrw4lp74h.
+ LOG.info(
+ "Merge the current non-terminated rescale and the new rescale triggered by recoverable failover into the current rescale.");
+ rescaleTimeline.updateRescale(Rescale::clearSchedulerStates);
+ } else if (rescaleTimeline.isIdling()) {
Review Comment:
This `else if (rescaleTimeline.isIdling())` branch is redundant because it’s the negation of the preceding `if (!rescaleTimeline.isIdling())`. Simplifying this improves readability and avoids suggesting different semantics between the two checks.
```suggestion
} else {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerstate();
Review Comment:
Method name `schedulerstate()` doesn’t follow the usual camelCase convention (should be `schedulerState()`), and the misspelling propagates into call sites. Renaming this now will improve readability and avoid the appearance of a typo in a core runtime interface.
```suggestion
State schedulerState();
```
##########
flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java:
##########
@@ -141,6 +141,20 @@ public class WebOptions {
.withDescription(
"The maximum number of failures collected by the exception history per job.");
+ /** The maximum number of the adaptive scheduler rescale history. */
+ @Documentation.Section(Documentation.Sections.ALL_JOB_MANAGER)
+ public static final ConfigOption<Integer> MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE =
+ key("web.adaptive-scheduler.rescale-history.size")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The maximum number of the rescale records per job whose scheduler is %s. "
+ + "The feature will be disabled when the configuration value is smaller or equals to 0.",
Review Comment:
Grammar in the config description: “smaller or equals to 0” should be “smaller than or equal to 0”.
```suggestion
+ "The feature will be disabled when the configuration value is smaller than or equal to 0.",
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -149,19 +175,31 @@ public ScheduledFuture<?> scheduleOperation(Runnable callback, Duration delay) {
@Override
public void transitionToSubsequentState() {
+ Optional<VertexParallelism> availableVertexParallelism =
+ context.getAvailableVertexParallelism();
+ if (!availableVertexParallelism.isPresent()) {
+ IllegalStateException exception =
+ new IllegalStateException("Resources must be available when rescaling.");
+ recordRescaleForNoResourcesEnough(exception);
+ throw exception;
+ }
context.goToRestarting(
getExecutionGraph(),
getExecutionGraphHandler(),
getOperatorCoordinatorHandler(),
Duration.ofMillis(0L),
- context.getAvailableVertexParallelism()
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Resources must be available when rescaling.")),
+ availableVertexParallelism.get(),
getFailures());
}
+ private void recordRescaleForNoResourcesEnough(IllegalStateException exception) {
+ context.getRescaleTimeline()
+ .updateRescale(
+ rescale ->
+ rescale.setStringifiedException(
+ ExceptionUtils.stringifyException(exception)));
Review Comment:
`recordRescaleForNoResourcesEnough` only stores the stringified exception, but it doesn’t terminate the rescale (no `TerminatedReason.EXCEPTION_OCCURRED` / end timestamp). Since the method throws immediately afterward, the current rescale will remain in-progress and may never be sealed, which can make rescale history inaccurate (and makes it hard to test the “resources not enough exception” termination case).
```suggestion
rescale -> {
rescale.setStringifiedException(
ExceptionUtils.stringifyException(exception));
rescale.setTerminatedReason(TerminatedReason.EXCEPTION_OCCURRED);
rescale.setEndTime(Instant.now());
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-3854652573
blocked #27538
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-3854640050
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "756fac008c343be88db156fd8b5e288f1dbdd368",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "756fac008c343be88db156fd8b5e288f1dbdd368",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 756fac008c343be88db156fd8b5e288f1dbdd368 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3007348820
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -95,18 +98,50 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ recordRescaleForJobIntoExecuting(logger, context);
+
deploy();
// check if new resources have come available in the meantime
context.runIfState(
this,
() -> {
- stateTransitionManager.onChange();
+ stateTransitionManager.onChange(true);
stateTransitionManager.onTrigger();
},
Duration.ZERO);
}
+ private void recordRescaleForJobIntoExecuting(Logger logger, Context context) {
+ // As described at https://github.com/apache/flink/pull/27539#discussion_r3006364815.
+ // If the transition is from the stop-with-savepoint state to the current state, a rescale
+ // record will be generated without any change in parallelism.
+ if (context.getState() instanceof StopWithSavepoint) {
+ logger.warn(
+ "The state switching is caused by {}->{}, the parallelisms would be not changed, so the rescale record is ignored.",
+ StopWithSavepoint.class.getSimpleName(),
+ Executing.class.getSimpleName());
+ } else {
+ context.getRescaleTimeline()
+ .updateRescale(
+ rescale ->
+ rescale.addSchedulerState(this)
+ .setTerminatedReason(TerminatedReason.SUCCEEDED)
+ .setEndToNow()
+ .log());
+ }
+ }
+
Review Comment:
anchor-A:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006350787
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
Thanks @ferenc-csaky
What about `RescaleTimelineContext` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal merged PR #27539:
URL: https://github.com/apache/flink/pull/27539
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006350787
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
Thanks @ferenc-csaky
It's reasonable .
And What about `RescaleContext` ?
If this proposal is not very suitable, then I will continue using the strategy you mentioned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-4150382043
Thanks @ferenc-csaky for your review.
I update the corresponding lines based on your comments except the `HasRescaleTimeline / RescaleContext` related part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal closed pull request #27539: [FLINK-38342][runtime] Support record rescale information.
URL: https://github.com/apache/flink/pull/27539
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006355437
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
Works for me, `RescaleContext` actually seems better, as we reuse it in other `Context` interfaces 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006384996
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
Thanks @ferenc-csaky .
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-4089400054
Hi, @X-czh Could you help take a look if you had the free time ?
Thank you ~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3007706626
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -95,18 +98,50 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ recordRescaleForJobIntoExecuting(logger, context);
+
deploy();
// check if new resources have come available in the meantime
context.runIfState(
this,
() -> {
- stateTransitionManager.onChange();
+ stateTransitionManager.onChange(true);
stateTransitionManager.onTrigger();
},
Duration.ZERO);
}
+ private void recordRescaleForJobIntoExecuting(Logger logger, Context context) {
+ // As described at https://github.com/apache/flink/pull/27539#discussion_r3006364815.
Review Comment:
I'd rather not reference the PR comment here, but instead make this comment contain the whole context, which is almost already true, maybe:
```
// When "AdaptiveScheduler -> stopWithSavepoint" fails, the job is not stopped, but
// instead transitions back to the "Executing" state. In this case, there is no
// parallelism change, hence we ignore it for rescale.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006384672
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -99,6 +98,7 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ // 如果不是从sp过来则记录，否则从sp过来则不记录
Review Comment:
Translated it into English now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006355437
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
Works for me 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3007874916
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -95,18 +98,50 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ recordRescaleForJobIntoExecuting(logger, context);
+
deploy();
// check if new resources have come available in the meantime
context.runIfState(
this,
() -> {
- stateTransitionManager.onChange();
+ stateTransitionManager.onChange(true);
stateTransitionManager.onTrigger();
},
Duration.ZERO);
}
+ private void recordRescaleForJobIntoExecuting(Logger logger, Context context) {
+ // As described at https://github.com/apache/flink/pull/27539#discussion_r3006364815.
Review Comment:
Nice comments.
Thanks @ferenc-csaky
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-4153745397
Thanks @ferenc-csaky for the review.
Merging...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #27539:
URL: https://github.com/apache/flink/pull/27539#issuecomment-3970197724
Hi, masters, @ferenc-csaky @davidradl
Could you help take a look for the PR if you had the free time ?
Thank you very much 🌹
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006204146
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1133,6 +1154,27 @@ public void updateJobResourceRequirements(JobResourceRequirements jobResourceReq
}
}
+ private void recordRescaleForNewResourceRequirements() {
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.addSchedulerState(state)
+ .setTerminatedReason(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED)
+ .setEndTimestamp(Instant.now().toEpochMilli())
+ .log());
+ rescaleTimeline.newRescale(true);
+ rescaleTimeline.updateRescale(
+ rescale ->
+ rescale.setStartTimestamp(Instant.now().toEpochMilli())
Review Comment:
I think it would worth adding a `setStartToNow()` and `setEndToNow()` that calls the setter for `Instant.now().toEpochMilli()` and use those when we can, cause most of the time this is what happens so we could shorten it a bit.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java:
##########
@@ -1336,13 +1389,55 @@ public void goToRestarting(
}
}
+ private void recordRescaleForJobRestarting(VertexParallelism restartWithParallelism) {
+ if (restartWithParallelism == null) {
Review Comment:
I'd probably invert this `if`, do the normal rescaling first, then return explicitly, so we can spare the `else` branch and the nested `if`, and 1 indentation level.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -149,19 +175,31 @@ public ScheduledFuture<?> scheduleOperation(Runnable callback, Duration delay) {
@Override
public void transitionToSubsequentState() {
+ Optional<VertexParallelism> availableVertexParallelism =
+ context.getAvailableVertexParallelism();
+ if (!availableVertexParallelism.isPresent()) {
Review Comment:
Use `availableVertexParallelism.isEmpty()` instead.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -49,6 +59,10 @@ default void close() {}
*/
interface Context {
+ State schedulerState();
+
+ RescaleTimeline getRescaleTimeline();
Review Comment:
I think I'd rather add a common interface (e.g. `HasRescaleTimeline`) that has `getRescaleTimeline` and then extend the `Context` with that interface here and in `StateWithExecutionGraph`, `StateWithoutExecutionGraph` as well, so we don't duplicate this method 3 times.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006364815
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -99,6 +98,7 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ // 如果不是从sp过来则记录，否则从sp过来则不记录
Review Comment:
Hi, @ferenc-csaky Thank you for the reminder.
To be exactly, It should be a `TODO` line in English.
With offline-discussion with @WeiZhong94 , We believe there may be some uncertainty and potential risks around this part of the code, so I’ve added a marker here.
Currently, in AdaptiveScheduler, when stopWithSavepoint fails, the job is not stopped but instead transitions back to the Executing state. This may result in an extra rescale record being generated without any actual change in parallelism.
I will make corresponding adjustments once we reach a clearer conclusion.
Of course, we would greatly appreciate your thoughts on this as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006357845
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -99,6 +98,7 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ // 如果不是从sp过来则记录，否则从sp过来则不记录
Review Comment:
If this is not some debug comment leftover, can we translate this to english?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38342][runtime] Support record rescale information. [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3006364815
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -99,6 +98,7 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ // 如果不是从sp过来则记录，否则从sp过来则不记录
Review Comment:
Hi, @ferenc-csaky Thank you for the reminder.
To be exactly, It should be a `TODO` line in English.
With offline-discussion with @WeiZhong94 , We believe there may be some uncertainty and potential risks around this part of the code, so I’ve added a marker here.
Currently, in AdaptiveScheduler, when stopWithSavepoint fails, the job is not stopped but instead transitions back to the Executing state. This may result in an extra rescale record being generated without any actual change in parallelism.
I will make corresponding adjustments once we reach a clearer conclusion.
Of course, we would greatly appreciate your thoughts on this as well.
A candidate solution is here https://github.com/apache/flink/pull/27539/changes#r3007348820
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org