You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/03/31 13:10:43 UTC
Re: [PR] [FLINK-38895][runtime] Introduce the /jobs/:jobid/rescales/details/:rescaleuuid endpoint in the REST API [flink]
Copilot commented on code in PR #27695:
URL: https://github.com/apache/flink/pull/27695#discussion_r3015813572
##########
flink-core/src/main/java/org/apache/flink/util/AbstractID.java:
##########
@@ -89,6 +89,11 @@ public AbstractID() {
this.upperPart = random.nextLong();
}
+ /** Constructs a new random ID from a uniform distribution. */
Review Comment:
The Javadoc for the new `AbstractID(String hexString)` constructor is incorrect: it says it constructs a random ID, but it actually parses an ID from the provided hex string. This is a public-evolving API, so the misleading documentation should be corrected to describe hex parsing (and ideally mention the expected length/format).
```suggestion
/**
* Constructs a new ID from its hexadecimal string representation.
*
* <p>The expected format is a 32-character lowercase hexadecimal string (16 bytes).
*
* @param hexString the hexadecimal string representation of the ID
* @throws IllegalArgumentException if the string is null, has an invalid length, or
* contains non-hex characters
*/
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/rescales/JobRescaleDetails.java:
##########
@@ -0,0 +1,508 @@
+/*
+ * 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.rest.messages.job.rescales;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.rest.messages.ResourceProfileInfo;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDKeyDeserializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDKeySerializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDSerializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDKeyDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDKeySerializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDSerializer;
+import org.apache.flink.runtime.scheduler.VertexParallelismInformation;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Rescale;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.SlotSharingGroupRescale;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TerminalState;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TerminatedReason;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TriggerCause;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Job rescales details class. */
+@Schema(name = "JobRescaleDetails")
+public class JobRescaleDetails implements ResponseBody, Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public static final String FIELD_NAME_RESCALE_UUID = "rescaleUuid";
+ public static final String FIELD_NAME_RESOURCE_REQUIREMENTS_UUID = "resourceRequirementsUuid";
+ public static final String FIELD_NAME_RESCALE_ATTEMPT_ID = "rescaleAttemptId";
+ public static final String FIELD_NAME_VERTICES = "vertices";
+ public static final String FIELD_NAME_SLOTS = "slots";
+ public static final String FIELD_NAME_SCHEDULER_STATES = "schedulerStates";
+ public static final String FIELD_NAME_START_TIMESTAMP = "startTimestampInMillis";
+ public static final String FIELD_NAME_END_TIMESTAMP = "endTimestampInMillis";
+ public static final String FIELD_NAME_TERMINAL_STATE = "terminalState";
+ public static final String FIELD_NAME_TRIGGER_CAUSE = "triggerCause";
+ public static final String FIELD_NAME_TERMINATED_REASON = "terminatedReason";
+
+ @JsonProperty(FIELD_NAME_RESCALE_UUID)
+ private final String rescaleUuid;
+
+ @JsonProperty(FIELD_NAME_RESOURCE_REQUIREMENTS_UUID)
+ private final String resourceRequirementsUuid;
+
+ @JsonProperty(FIELD_NAME_RESCALE_ATTEMPT_ID)
+ private final long rescaleAttemptId;
+
+ @JsonProperty(FIELD_NAME_VERTICES)
+ @JsonSerialize(keyUsing = JobVertexIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
+ private final Map<JobVertexID, VertexParallelismRescaleInfo> vertices;
+
+ @JsonProperty(FIELD_NAME_SLOTS)
+ @JsonSerialize(keyUsing = SlotSharingGroupIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = SlotSharingGroupIDKeyDeserializer.class)
+ private final Map<SlotSharingGroupId, SlotSharingGroupRescaleInfo> slots;
+
+ @JsonProperty(FIELD_NAME_SCHEDULER_STATES)
+ private final List<SchedulerStateSpan> schedulerStates;
+
+ @JsonProperty(FIELD_NAME_START_TIMESTAMP)
+ private final Long startTimestamp;
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_END_TIMESTAMP)
+ private final Long endTimestamp;
+
+ @JsonProperty(FIELD_NAME_TERMINAL_STATE)
+ private final TerminalState terminalState;
+
+ @JsonProperty(FIELD_NAME_TRIGGER_CAUSE)
+ private final TriggerCause triggerCause;
+
+ @JsonProperty(FIELD_NAME_TERMINATED_REASON)
+ private final TerminatedReason terminatedReason;
+
+ @JsonCreator
+ public JobRescaleDetails(
+ @JsonProperty(FIELD_NAME_RESCALE_UUID) String rescaleUuid,
+ @JsonProperty(FIELD_NAME_RESOURCE_REQUIREMENTS_UUID) String resourceRequirementsUuid,
+ @JsonProperty(FIELD_NAME_RESCALE_ATTEMPT_ID) long rescaleAttemptId,
+ @JsonSerialize(keyUsing = JobVertexIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
+ @JsonProperty(FIELD_NAME_VERTICES)
+ Map<JobVertexID, VertexParallelismRescaleInfo> vertices,
+ @JsonSerialize(keyUsing = SlotSharingGroupIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = SlotSharingGroupIDKeyDeserializer.class)
+ @JsonProperty(FIELD_NAME_SLOTS)
+ Map<SlotSharingGroupId, SlotSharingGroupRescaleInfo> slots,
+ @JsonProperty(FIELD_NAME_SCHEDULER_STATES) List<SchedulerStateSpan> schedulerStates,
+ @JsonProperty(FIELD_NAME_START_TIMESTAMP) Long startTimestamp,
+ @Nullable @JsonProperty(FIELD_NAME_END_TIMESTAMP) Long endTimestamp,
+ @JsonProperty(FIELD_NAME_TERMINAL_STATE) TerminalState terminalState,
+ @JsonProperty(FIELD_NAME_TRIGGER_CAUSE) TriggerCause triggerCause,
+ @JsonProperty(FIELD_NAME_TERMINATED_REASON) TerminatedReason terminatedReason) {
+ this.rescaleUuid = rescaleUuid;
+ this.resourceRequirementsUuid = resourceRequirementsUuid;
+ this.rescaleAttemptId = rescaleAttemptId;
+ this.vertices = vertices;
+ this.slots = slots;
+ this.schedulerStates = schedulerStates;
+ this.startTimestamp = startTimestamp;
+ this.endTimestamp = endTimestamp;
+ this.terminalState = terminalState;
+ this.triggerCause = triggerCause;
+ this.terminatedReason = terminatedReason;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ JobRescaleDetails that = (JobRescaleDetails) o;
+ return rescaleAttemptId == that.rescaleAttemptId
+ && Objects.equals(rescaleUuid, that.rescaleUuid)
+ && Objects.equals(resourceRequirementsUuid, that.resourceRequirementsUuid)
+ && Objects.equals(vertices, that.vertices)
+ && Objects.equals(slots, that.slots)
+ && Objects.equals(schedulerStates, that.schedulerStates)
+ && Objects.equals(startTimestamp, that.startTimestamp)
+ && Objects.equals(endTimestamp, that.endTimestamp)
+ && terminalState == that.terminalState
+ && triggerCause == that.triggerCause
+ && terminatedReason == that.terminatedReason;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ rescaleUuid,
+ resourceRequirementsUuid,
+ rescaleAttemptId,
+ vertices,
+ slots,
+ schedulerStates,
+ startTimestamp,
+ endTimestamp,
+ terminalState,
+ triggerCause,
+ terminatedReason);
+ }
+
+ public static JobRescaleDetails fromRescale(Rescale rescale, boolean includeSchedulerStates) {
+ return new JobRescaleDetails(
+ rescale.getRescaleIdInfo().getRescaleUuid().toString(),
+ rescale.getRescaleIdInfo().getResourceRequirementsId().toString(),
+ rescale.getRescaleIdInfo().getRescaleAttemptId(),
+ rescale.getVertices(),
+ convertMapValues(
+ rescale.getSlots(),
+ SlotSharingGroupRescaleInfo::fromSlotSharingGroupRescale),
+ includeSchedulerStates ? rescale.getSchedulerStates() : null,
+ rescale.getStartTimestamp(),
+ rescale.getEndTimestamp(),
+ rescale.getTerminalState(),
+ rescale.getTriggerCause(),
+ rescale.getTerminatedReason());
Review Comment:
`fromRescale(..., includeSchedulerStates)` passes `null` for `schedulerStates` when `includeSchedulerStates` is false, but the field is modeled as a list/array in the REST schema. Returning `null` here can lead to inconsistent payloads and NPE-prone client code. Prefer returning an empty list (or using `@JsonInclude(NON_NULL)` and marking the field as nullable if `null` is intended).
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/rescales/JobRescaleIDPathParameter.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.rest.messages.job.rescales;
+
+import org.apache.flink.runtime.rest.messages.ConversionException;
+import org.apache.flink.runtime.rest.messages.MessagePathParameter;
+import org.apache.flink.util.AbstractID;
+
+/** The {@link MessagePathParameter} for {@link JobRescaleDetailsHeaders}. */
+public class JobRescaleIDPathParameter extends MessagePathParameter<AbstractID> {
+
+ public static final String KEY = "rescaleuuid";
+
+ protected JobRescaleIDPathParameter() {
+ super(KEY);
+ }
+
+ @Override
+ protected AbstractID convertFromString(String value) throws ConversionException {
+ try {
+ return new AbstractID(value);
+ } catch (NumberFormatException nfe) {
+ throw new ConversionException("Could not parse long from " + value + '.', nfe);
Review Comment:
`convertFromString` is catching `NumberFormatException` and rethrowing a `ConversionException` with message "Could not parse long...", but `new AbstractID(value)` ultimately delegates to hex parsing and is more likely to throw `IllegalArgumentException` (or similar) for invalid input. As-is, invalid `rescaleuuid` values may escape as unchecked exceptions and result in 500s, and the error message is misleading. Please catch the appropriate exception type(s) and use an error message that refers to parsing an AbstractID/rescale UUID from a hex string; also consider using a dedicated ID type (similar to `TriggerId`) for proper OpenAPI schema generation instead of exposing `AbstractID` as an object in the docs.
```suggestion
} catch (IllegalArgumentException | NumberFormatException ex) {
throw new ConversionException(
"Could not parse rescale UUID (AbstractID) from hex string: " + value + '.',
ex);
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/rescales/JobRescaleDetails.java:
##########
@@ -0,0 +1,508 @@
+/*
+ * 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.rest.messages.job.rescales;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.rest.messages.ResourceProfileInfo;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDKeyDeserializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDKeySerializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDSerializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDKeyDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDKeySerializer;
+import org.apache.flink.runtime.rest.messages.json.SlotSharingGroupIDSerializer;
+import org.apache.flink.runtime.scheduler.VertexParallelismInformation;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Rescale;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.SlotSharingGroupRescale;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TerminalState;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TerminatedReason;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.TriggerCause;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Job rescales details class. */
+@Schema(name = "JobRescaleDetails")
+public class JobRescaleDetails implements ResponseBody, Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public static final String FIELD_NAME_RESCALE_UUID = "rescaleUuid";
+ public static final String FIELD_NAME_RESOURCE_REQUIREMENTS_UUID = "resourceRequirementsUuid";
+ public static final String FIELD_NAME_RESCALE_ATTEMPT_ID = "rescaleAttemptId";
+ public static final String FIELD_NAME_VERTICES = "vertices";
+ public static final String FIELD_NAME_SLOTS = "slots";
+ public static final String FIELD_NAME_SCHEDULER_STATES = "schedulerStates";
+ public static final String FIELD_NAME_START_TIMESTAMP = "startTimestampInMillis";
+ public static final String FIELD_NAME_END_TIMESTAMP = "endTimestampInMillis";
+ public static final String FIELD_NAME_TERMINAL_STATE = "terminalState";
+ public static final String FIELD_NAME_TRIGGER_CAUSE = "triggerCause";
+ public static final String FIELD_NAME_TERMINATED_REASON = "terminatedReason";
+
+ @JsonProperty(FIELD_NAME_RESCALE_UUID)
+ private final String rescaleUuid;
+
+ @JsonProperty(FIELD_NAME_RESOURCE_REQUIREMENTS_UUID)
+ private final String resourceRequirementsUuid;
+
+ @JsonProperty(FIELD_NAME_RESCALE_ATTEMPT_ID)
+ private final long rescaleAttemptId;
+
+ @JsonProperty(FIELD_NAME_VERTICES)
+ @JsonSerialize(keyUsing = JobVertexIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
+ private final Map<JobVertexID, VertexParallelismRescaleInfo> vertices;
+
+ @JsonProperty(FIELD_NAME_SLOTS)
+ @JsonSerialize(keyUsing = SlotSharingGroupIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = SlotSharingGroupIDKeyDeserializer.class)
+ private final Map<SlotSharingGroupId, SlotSharingGroupRescaleInfo> slots;
+
+ @JsonProperty(FIELD_NAME_SCHEDULER_STATES)
+ private final List<SchedulerStateSpan> schedulerStates;
+
+ @JsonProperty(FIELD_NAME_START_TIMESTAMP)
+ private final Long startTimestamp;
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_END_TIMESTAMP)
+ private final Long endTimestamp;
+
+ @JsonProperty(FIELD_NAME_TERMINAL_STATE)
+ private final TerminalState terminalState;
+
+ @JsonProperty(FIELD_NAME_TRIGGER_CAUSE)
+ private final TriggerCause triggerCause;
+
+ @JsonProperty(FIELD_NAME_TERMINATED_REASON)
+ private final TerminatedReason terminatedReason;
+
+ @JsonCreator
+ public JobRescaleDetails(
+ @JsonProperty(FIELD_NAME_RESCALE_UUID) String rescaleUuid,
+ @JsonProperty(FIELD_NAME_RESOURCE_REQUIREMENTS_UUID) String resourceRequirementsUuid,
+ @JsonProperty(FIELD_NAME_RESCALE_ATTEMPT_ID) long rescaleAttemptId,
+ @JsonSerialize(keyUsing = JobVertexIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
+ @JsonProperty(FIELD_NAME_VERTICES)
+ Map<JobVertexID, VertexParallelismRescaleInfo> vertices,
+ @JsonSerialize(keyUsing = SlotSharingGroupIDKeySerializer.class)
+ @JsonDeserialize(keyUsing = SlotSharingGroupIDKeyDeserializer.class)
+ @JsonProperty(FIELD_NAME_SLOTS)
+ Map<SlotSharingGroupId, SlotSharingGroupRescaleInfo> slots,
+ @JsonProperty(FIELD_NAME_SCHEDULER_STATES) List<SchedulerStateSpan> schedulerStates,
+ @JsonProperty(FIELD_NAME_START_TIMESTAMP) Long startTimestamp,
+ @Nullable @JsonProperty(FIELD_NAME_END_TIMESTAMP) Long endTimestamp,
+ @JsonProperty(FIELD_NAME_TERMINAL_STATE) TerminalState terminalState,
+ @JsonProperty(FIELD_NAME_TRIGGER_CAUSE) TriggerCause triggerCause,
+ @JsonProperty(FIELD_NAME_TERMINATED_REASON) TerminatedReason terminatedReason) {
+ this.rescaleUuid = rescaleUuid;
+ this.resourceRequirementsUuid = resourceRequirementsUuid;
+ this.rescaleAttemptId = rescaleAttemptId;
+ this.vertices = vertices;
+ this.slots = slots;
+ this.schedulerStates = schedulerStates;
+ this.startTimestamp = startTimestamp;
+ this.endTimestamp = endTimestamp;
+ this.terminalState = terminalState;
+ this.triggerCause = triggerCause;
+ this.terminatedReason = terminatedReason;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ JobRescaleDetails that = (JobRescaleDetails) o;
+ return rescaleAttemptId == that.rescaleAttemptId
+ && Objects.equals(rescaleUuid, that.rescaleUuid)
+ && Objects.equals(resourceRequirementsUuid, that.resourceRequirementsUuid)
+ && Objects.equals(vertices, that.vertices)
+ && Objects.equals(slots, that.slots)
+ && Objects.equals(schedulerStates, that.schedulerStates)
+ && Objects.equals(startTimestamp, that.startTimestamp)
+ && Objects.equals(endTimestamp, that.endTimestamp)
+ && terminalState == that.terminalState
+ && triggerCause == that.triggerCause
+ && terminatedReason == that.terminatedReason;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ rescaleUuid,
+ resourceRequirementsUuid,
+ rescaleAttemptId,
+ vertices,
+ slots,
+ schedulerStates,
+ startTimestamp,
+ endTimestamp,
+ terminalState,
+ triggerCause,
+ terminatedReason);
+ }
+
+ public static JobRescaleDetails fromRescale(Rescale rescale, boolean includeSchedulerStates) {
+ return new JobRescaleDetails(
+ rescale.getRescaleIdInfo().getRescaleUuid().toString(),
+ rescale.getRescaleIdInfo().getResourceRequirementsId().toString(),
+ rescale.getRescaleIdInfo().getRescaleAttemptId(),
+ rescale.getVertices(),
+ convertMapValues(
+ rescale.getSlots(),
+ SlotSharingGroupRescaleInfo::fromSlotSharingGroupRescale),
+ includeSchedulerStates ? rescale.getSchedulerStates() : null,
+ rescale.getStartTimestamp(),
+ rescale.getEndTimestamp(),
+ rescale.getTerminalState(),
+ rescale.getTriggerCause(),
+ rescale.getTerminatedReason());
+ }
+
+ private static <K, NV, OV> Map<K, NV> convertMapValues(
+ Map<K, OV> rawMap, Function<OV, NV> valueMapper) {
+ return rawMap == null
+ ? new HashMap<>()
+ : rawMap.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ kovEntry -> valueMapper.apply(kovEntry.getValue())));
+ }
+
+ /** The rescale information of a {@link org.apache.flink.runtime.jobgraph.JobVertex}. */
+ public static final class VertexParallelismRescaleInfo implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public static final String FIELD_NAME_JOB_VERTEX_ID = "jobVertexId";
+ public static final String FIELD_NAME_VERTEX_NAME = "jobVertexName";
+ public static final String FIELD_NAME_SLOT_SHARING_GROUP_ID = "slotSharingGroupId";
+ public static final String FIELD_NAME_SLOT_SHARING_GROUP_NAME = "slotSharingGroupName";
+ public static final String FIELD_NAME_DESIRED_PARALLELISM = "desiredParallelism";
+ public static final String FIELD_NAME_SUFFICIENT_PARALLELISM = "sufficientParallelism";
+ public static final String FIELD_NAME_PRE_RESCALE_PARALLELISM = "preRescaleParallelism";
+ public static final String FIELD_NAME_POST_RESCALE_PARALLELISM = "postRescaleParallelism";
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_ID)
+ @JsonSerialize(using = JobVertexIDSerializer.class)
+ @JsonDeserialize(using = JobVertexIDDeserializer.class)
+ private final JobVertexID jobVertexId;
+
+ @JsonProperty(FIELD_NAME_VERTEX_NAME)
+ private String jobVertexName;
+
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_ID)
+ @JsonSerialize(using = SlotSharingGroupIDSerializer.class)
+ @JsonDeserialize(using = SlotSharingGroupIDDeserializer.class)
+ private SlotSharingGroupId slotSharingGroupId;
+
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_NAME)
+ private String slotSharingGroupName;
+
+ @JsonProperty(FIELD_NAME_DESIRED_PARALLELISM)
+ private Integer desiredParallelism;
+
+ @JsonProperty(FIELD_NAME_SUFFICIENT_PARALLELISM)
+ private Integer sufficientParallelism;
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_PRE_RESCALE_PARALLELISM)
+ private Integer preRescaleParallelism;
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_POST_RESCALE_PARALLELISM)
+ private Integer postRescaleParallelism;
+
+ @JsonCreator
+ public VertexParallelismRescaleInfo(
+ @JsonSerialize(using = JobVertexIDSerializer.class)
+ @JsonDeserialize(using = JobVertexIDDeserializer.class)
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_ID)
+ JobVertexID jobVertexId,
+ @JsonProperty(FIELD_NAME_VERTEX_NAME) String jobVertexName,
+ @JsonSerialize(using = SlotSharingGroupIDSerializer.class)
+ @JsonDeserialize(using = SlotSharingGroupIDDeserializer.class)
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_ID)
+ SlotSharingGroupId slotSharingGroupId,
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_NAME) String slotSharingGroupName,
+ @JsonProperty(FIELD_NAME_DESIRED_PARALLELISM) Integer desiredParallelism,
+ @JsonProperty(FIELD_NAME_SUFFICIENT_PARALLELISM) Integer sufficientParallelism,
+ @JsonProperty(FIELD_NAME_PRE_RESCALE_PARALLELISM) Integer preRescaleParallelism,
+ @Nullable @JsonProperty(FIELD_NAME_POST_RESCALE_PARALLELISM)
+ Integer postRescaleParallelism) {
+ this.jobVertexId = jobVertexId;
+ this.jobVertexName = jobVertexName;
+ this.slotSharingGroupId = slotSharingGroupId;
+ this.slotSharingGroupName = slotSharingGroupName;
+ this.desiredParallelism = desiredParallelism;
+ this.sufficientParallelism = sufficientParallelism;
+ this.preRescaleParallelism = preRescaleParallelism;
+ this.postRescaleParallelism = postRescaleParallelism;
+ }
+
+ @JsonIgnore
+ public VertexParallelismRescaleInfo(
+ JobVertexID jobVertexId, String jobVertexName, SlotSharingGroup slotSharingGroup) {
+ this.jobVertexId = Preconditions.checkNotNull(jobVertexId);
+ this.jobVertexName = jobVertexName;
+ this.slotSharingGroupName = slotSharingGroup.getSlotSharingGroupName();
+ this.slotSharingGroupId = slotSharingGroup.getSlotSharingGroupId();
+ }
+
+ public JobVertexID getJobVertexId() {
+ return jobVertexId;
+ }
+
+ public String getJobVertexName() {
+ return jobVertexName;
+ }
+
+ public void setJobVertexName(String jobVertexName) {
+ this.jobVertexName = jobVertexName;
+ }
+
+ public SlotSharingGroupId getSlotSharingGroupId() {
+ return slotSharingGroupId;
+ }
+
+ public String getSlotSharingGroupName() {
+ return slotSharingGroupName;
+ }
+
+ @Nullable
+ public Integer getPreRescaleParallelism() {
+ return preRescaleParallelism;
+ }
+
+ public void setPreRescaleParallelism(@Nullable Integer preRescaleParallelism) {
+ this.preRescaleParallelism = preRescaleParallelism;
+ }
+
+ public Integer getDesiredParallelism() {
+ return desiredParallelism;
+ }
+
+ public Integer getSufficientParallelism() {
+ return sufficientParallelism;
+ }
+
+ public void setRequiredParallelisms(
+ VertexParallelismInformation vertexParallelismInformation) {
+ this.sufficientParallelism = vertexParallelismInformation.getMinParallelism();
+ this.desiredParallelism = vertexParallelismInformation.getParallelism();
+ }
Review Comment:
`VertexParallelismRescaleInfo#setRequiredParallelisms(...)` is a public JavaBean-style setter that does not correspond to any serialized field. This leaks into generated OpenAPI docs as a `requiredParallelisms` property (see the updated `rest_v1_dispatcher.yml`), even though it will never be present in actual JSON responses. Consider renaming it to a non-bean method name (e.g., `applyRequiredParallelisms`) and/or annotating it with `@JsonIgnore`/`@Schema(hidden = true)` to keep the public REST schema accurate.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/Rescale.java:
##########
@@ -134,15 +136,15 @@ public class Rescale implements Serializable {
@Nullable private TerminalState terminalState;
@Nullable private TerminatedReason terminatedReason;
- Rescale(RescaleIdInfo rescaleIdInfo) {
+ public Rescale(RescaleIdInfo rescaleIdInfo) {
this.rescaleIdInfo = rescaleIdInfo;
this.vertices = new HashMap<>();
this.slots = new HashMap<>();
this.schedulerStates = new ArrayList<>();
}
@VisibleForTesting
- Rescale addSchedulerState(SchedulerStateSpan schedulerStateSpan) {
+ public Rescale addSchedulerState(SchedulerStateSpan schedulerStateSpan) {
Review Comment:
Several members were made `public` (constructor, `addSchedulerState(SchedulerStateSpan)`, and `getModifiableVertices()/getModifiableSlots()`) even though they are annotated/used for testing. This widens the production API surface of `Rescale` and allows external code to mutate internal state. If the intent is test-only access, consider keeping these package-private and relocating tests, or provide narrowly-scoped `@VisibleForTesting` helpers instead of making the mutators publicly accessible.
```suggestion
Rescale addSchedulerState(SchedulerStateSpan schedulerStateSpan) {
```
--
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