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 fb4a6a8d60f [FLINK-40969][runtime] Fix rescale details lookup for
evicted latest events.
fb4a6a8d60f is described below
commit fb4a6a8d60fe9b8a4a4affc41eea57451de79c0a
Author: danzhewuju <[email protected]>
AuthorDate: Mon Oct 12 18:03:24 2026 +0800
[FLINK-40969][runtime] Fix rescale details lookup for evicted latest events.
Signed-off-by: danzhewuju <[email protected]>
Generated-by: GPT-6
---
.../job/rescales/JobRescaleDetailsHandler.java | 8 +-
.../adaptive/timeline/RescalesStatsSnapshot.java | 21 +--
.../job/rescales/JobRescaleDetailsHandlerTest.java | 143 +++++++++++++++++++++
3 files changed, 160 insertions(+), 12 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandler.java
index d3f94a32c24..5d39bcf0f0b 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandler.java
@@ -101,20 +101,20 @@ public class JobRescaleDetailsHandler
@Override
public Collection<ArchivedJson> archiveJsonWithPath(ExecutionGraphInfo
executionGraphInfo)
throws IOException {
- if (executionGraphInfo.getRescalesStatsSnapshot() == null
- ||
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory() == null) {
+ final RescalesStatsSnapshot rescalesStatsSnapshot =
+ executionGraphInfo.getRescalesStatsSnapshot();
+ if (rescalesStatsSnapshot == null) {
return List.of();
}
List<ArchivedJson> archives = new ArrayList<>();
- List<Rescale> rescales =
executionGraphInfo.getRescalesStatsSnapshot().getRescaleHistory();
final String commonPathPrefix =
JobRescaleDetailsHeaders.getInstance()
.getTargetRestEndpointURL()
.replace(
':' + JobIDPathParameter.KEY,
executionGraphInfo.getJobId().toString());
- for (Rescale rescale : rescales) {
+ for (Rescale rescale : rescalesStatsSnapshot.getRescales()) {
archives.add(
new ArchivedJson(
commonPathPrefix.replace(
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescalesStatsSnapshot.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescalesStatsSnapshot.java
index 0dfd3fed380..109bd0f4eda 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescalesStatsSnapshot.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescalesStatsSnapshot.java
@@ -25,11 +25,10 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
public class RescalesStatsSnapshot implements Serializable {
private static final long serialVersionUID = 1L;
@@ -44,13 +43,15 @@ public class RescalesStatsSnapshot implements Serializable {
Map<TerminalState, Rescale> latestRescales,
RescalesSummarySnapshot rescalesSummarySnapshot) {
this.rescaleHistory = List.copyOf(rescaleHistory);
- this.idToRescaleMap =
- rescaleHistory.stream()
- .collect(
- Collectors.toMap(
- r ->
r.getRescaleIdInfo().getRescaleUuid(),
- Function.identity()));
this.latestRescales = Map.copyOf(latestRescales);
+ final Map<AbstractID, Rescale> rescalesById = new HashMap<>();
+ for (Rescale rescale : rescaleHistory) {
+ rescalesById.put(rescale.getRescaleIdInfo().getRescaleUuid(),
rescale);
+ }
+ for (Rescale rescale : latestRescales.values()) {
+
rescalesById.putIfAbsent(rescale.getRescaleIdInfo().getRescaleUuid(), rescale);
+ }
+ this.idToRescaleMap = Map.copyOf(rescalesById);
this.rescalesSummarySnapshot = rescalesSummarySnapshot;
}
@@ -62,6 +63,10 @@ public class RescalesStatsSnapshot implements Serializable {
return idToRescaleMap.get(rescaleId);
}
+ public Collection<Rescale> getRescales() {
+ return idToRescaleMap.values();
+ }
+
@Nullable
public Rescale getLatestRescale(TerminalState terminalState) {
return latestRescales.get(terminalState);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandlerTest.java
index be0275a42b6..4b0e129c247 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/rescales/JobRescaleDetailsHandlerTest.java
@@ -44,11 +44,16 @@ import
org.apache.flink.runtime.scheduler.adaptive.timeline.RescalesSummary;
import
org.apache.flink.runtime.scheduler.adaptive.timeline.SlotSharingGroupRescale;
import org.apache.flink.runtime.scheduler.adaptive.timeline.TerminatedReason;
import org.apache.flink.runtime.scheduler.adaptive.timeline.TriggerCause;
+import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.util.concurrent.Executors;
+import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
import org.junit.jupiter.api.Test;
+import java.io.IOException;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -160,6 +165,144 @@ class JobRescaleDetailsHandlerTest {
assertThat(jobRescaleDetails).isEqualTo(fromRescale(rescale, true));
}
+ @Test
+ void testRequestLatestRescaleEvictedFromHistory()
+ throws HandlerRequestException, RestHandlerException {
+ final Rescale latestIgnoredRescale =
+
createTerminatedRescale(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
+ final Rescale retainedCompletedRescale =
+ createTerminatedRescale(TerminatedReason.SUCCEEDED);
+ final ExecutionGraphInfo executionGraphInfo =
+ createExecutionGraphInfo(latestIgnoredRescale,
retainedCompletedRescale);
+ final HandlerRequest<EmptyRequestBody> request =
+ createRequest(
+ executionGraphInfo.getJobId(),
+
latestIgnoredRescale.getRescaleIdInfo().getRescaleUuid());
+
+ assertThat(testInstance.handleRequest(request, executionGraphInfo))
+ .isEqualTo(fromRescale(latestIgnoredRescale, true));
+ }
+
+ @Test
+ void testRequestHistoryOnlyRescale() throws HandlerRequestException,
RestHandlerException {
+ final Rescale latestIgnoredRescale =
+
createTerminatedRescale(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
+ final Rescale retainedCompletedRescale =
+ createTerminatedRescale(TerminatedReason.SUCCEEDED);
+ final Rescale historyOnlyFailedRescale =
+ createTerminatedRescale(TerminatedReason.EXCEPTION_OCCURRED);
+ final ExecutionGraphInfo executionGraphInfo =
+ createExecutionGraphInfo(
+ latestIgnoredRescale,
+ retainedCompletedRescale,
+ List.of(historyOnlyFailedRescale,
retainedCompletedRescale));
+ final HandlerRequest<EmptyRequestBody> request =
+ createRequest(
+ executionGraphInfo.getJobId(),
+
historyOnlyFailedRescale.getRescaleIdInfo().getRescaleUuid());
+
+ assertThat(testInstance.handleRequest(request, executionGraphInfo))
+ .isEqualTo(fromRescale(historyOnlyFailedRescale, true));
+ }
+
+ @Test
+ void testRequestUnknownRescale() throws HandlerRequestException {
+ final Rescale latestIgnoredRescale =
+
createTerminatedRescale(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
+ final Rescale retainedCompletedRescale =
+ createTerminatedRescale(TerminatedReason.SUCCEEDED);
+ final Rescale historyOnlyFailedRescale =
+ createTerminatedRescale(TerminatedReason.EXCEPTION_OCCURRED);
+ final ExecutionGraphInfo executionGraphInfo =
+ createExecutionGraphInfo(
+ latestIgnoredRescale,
+ retainedCompletedRescale,
+ List.of(historyOnlyFailedRescale,
retainedCompletedRescale));
+ final RescaleIdInfo.RescaleUUID unknownRescaleUuid = new
RescaleIdInfo.RescaleUUID();
+ final HandlerRequest<EmptyRequestBody> request =
+ createRequest(executionGraphInfo.getJobId(),
unknownRescaleUuid);
+
+ assertThatThrownBy(() -> testInstance.handleRequest(request,
executionGraphInfo))
+ .isInstanceOfSatisfying(
+ RestHandlerException.class,
+ exception -> {
+ assertThat(exception.getHttpResponseStatus())
+ .isEqualTo(HttpResponseStatus.NOT_FOUND);
+ assertThat(exception.getMessage())
+ .contains(unknownRescaleUuid.toString());
+ });
+ }
+
+ @Test
+ void testArchiveIncludesLatestRescaleEvictedFromHistory() throws
IOException {
+ final Rescale latestIgnoredRescale =
+
createTerminatedRescale(TerminatedReason.RESOURCE_REQUIREMENTS_UPDATED);
+ final Rescale retainedCompletedRescale =
+ createTerminatedRescale(TerminatedReason.SUCCEEDED);
+ final Rescale historyOnlyFailedRescale =
+ createTerminatedRescale(TerminatedReason.EXCEPTION_OCCURRED);
+ final ExecutionGraphInfo executionGraphInfo =
+ createExecutionGraphInfo(
+ latestIgnoredRescale,
+ retainedCompletedRescale,
+ List.of(historyOnlyFailedRescale,
retainedCompletedRescale));
+
+ final Collection<ArchivedJson> archives =
+ testInstance.archiveJsonWithPath(executionGraphInfo);
+
+ assertThat(archives)
+ .extracting(ArchivedJson::getPath)
+ .containsExactlyInAnyOrder(
+ getDetailsPath(executionGraphInfo,
latestIgnoredRescale),
+ getDetailsPath(executionGraphInfo,
retainedCompletedRescale),
+ getDetailsPath(executionGraphInfo,
historyOnlyFailedRescale));
+ }
+
+ private static ExecutionGraphInfo createExecutionGraphInfo(
+ Rescale latestIgnoredRescale, Rescale retainedCompletedRescale) {
+ return createExecutionGraphInfo(
+ latestIgnoredRescale, retainedCompletedRescale,
List.of(retainedCompletedRescale));
+ }
+
+ private static ExecutionGraphInfo createExecutionGraphInfo(
+ Rescale latestIgnoredRescale,
+ Rescale retainedCompletedRescale,
+ List<Rescale> rescaleHistory) {
+ final RescalesSummary rescalesSummary = new RescalesSummary(1);
+ rescalesSummary.addTerminated(latestIgnoredRescale);
+ rescalesSummary.addTerminated(retainedCompletedRescale);
+ return new ExecutionGraphInfo(
+ new ArchivedExecutionGraphBuilder().build(),
+ List.of(),
+ null,
+ null,
+ new RescalesStatsSnapshot(
+ rescaleHistory,
+ Map.of(
+ latestIgnoredRescale.getTerminalState(),
+ latestIgnoredRescale,
+ retainedCompletedRescale.getTerminalState(),
+ retainedCompletedRescale),
+ rescalesSummary.createSnapshot()));
+ }
+
+ private static Rescale createTerminatedRescale(TerminatedReason
terminatedReason) {
+ return new Rescale(new RescaleIdInfo(new
RescaleIdInfo.ResourceRequirementsID(), 1L))
+ .setStartTimestamp(1L)
+ .setEndTimestamp(2L)
+ .setTriggerCause(TriggerCause.UPDATE_REQUIREMENT)
+ .setTerminatedReason(terminatedReason);
+ }
+
+ private static String getDetailsPath(ExecutionGraphInfo
executionGraphInfo, Rescale rescale) {
+ return JobRescaleDetailsHeaders.getInstance()
+ .getTargetRestEndpointURL()
+ .replace(':' + JobIDPathParameter.KEY,
executionGraphInfo.getJobId().toString())
+ .replace(
+ ':' + JobRescaleIDPathParameter.KEY,
+
rescale.getRescaleIdInfo().getRescaleUuid().toString());
+ }
+
private static HandlerRequest<EmptyRequestBody> createRequest(
JobID jobId, RescaleIdInfo.RescaleUUID rescaleUuid) throws
HandlerRequestException {
final Map<String, String> pathParameters = new HashMap<>();
