This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new baad9008 [FLINK-34574] Add CPU and memory size autoscaler quota
baad9008 is described below
commit baad90088ea5b5b240186a530a79b64fb84cc77e
Author: Gabor Somogyi <[email protected]>
AuthorDate: Fri Apr 19 17:36:24 2024 +0200
[FLINK-34574] Add CPU and memory size autoscaler quota
---
.../generated/auto_scaler_configuration.html | 12 +
.../apache/flink/autoscaler/ScalingExecutor.java | 113 +++++-
.../flink/autoscaler/ScalingMetricCollector.java | 15 +-
.../flink/autoscaler/config/AutoScalerOptions.java | 17 +
.../flink/autoscaler/topology/JobTopology.java | 18 +-
.../flink/autoscaler/topology/VertexInfo.java | 23 +-
.../runtime/rest/messages/job/JobDetailsInfo.java | 448 +++++++++++++++++++++
.../json/SlotSharingGroupIDDeserializer.java | 44 ++
.../json/SlotSharingGroupIDSerializer.java | 43 ++
.../messages/json/SlotSharingGroupIdConverter.java | 42 ++
.../MetricsCollectionAndEvaluationTest.java | 11 +-
.../flink/autoscaler/ScalingExecutorTest.java | 193 ++++++++-
.../autoscaler/ScalingMetricCollectorTest.java | 10 +
.../flink/autoscaler/TestingAutoscalerUtils.java | 9 +-
.../flink/autoscaler/topology/JobTopologyTest.java | 2 +-
.../operator/config/FlinkConfigBuilder.java | 18 +-
.../kubernetes/operator/utils/EventRecorder.java | 1 +
.../operator/config/FlinkConfigBuilderTest.java | 4 +
18 files changed, 977 insertions(+), 46 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index b3466b63..d9bed101 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
@@ -116,6 +116,18 @@
<td>Double</td>
<td>Percentage threshold for switching to observed from busy time
based true processing rate if the measurement is off by at least the configured
fraction. For example 0.15 means we switch to observed if the busy time based
computation is at least 15% higher during catchup.</td>
</tr>
+ <tr>
+ <td><h5>job.autoscaler.quota.cpu</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Double</td>
+ <td>Quota of the CPU count. When scaling would go beyond this
number the the scaling is not going to happen.</td>
+ </tr>
+ <tr>
+ <td><h5>job.autoscaler.quota.memory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Quota of the memory size. When scaling would go beyond this
number the the scaling is not going to happen.</td>
+ </tr>
<tr>
<td><h5>job.autoscaler.restart.time</h5></td>
<td style="word-wrap: break-word;">5 min</td>
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
index 71efc160..af325371 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
@@ -33,6 +33,7 @@ import org.apache.flink.autoscaler.utils.ResourceCheckUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.slf4j.Logger;
@@ -67,6 +68,9 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
public static final String HEAP_USAGE_MESSAGE =
"Heap Usage %s is above the allowed limit for scaling operations.
Please adjust the available memory manually.";
+ public static final String RESOURCE_QUOTA_REACHED_MESSAGE =
+ "Resource usage is above the allowed limit for scaling operations.
Please adjust the resource quota manually.";
+
private static final Logger LOG =
LoggerFactory.getLogger(ScalingExecutor.class);
private final JobVertexScaler<KEY, Context> jobVertexScaler;
@@ -129,8 +133,10 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
scalingSummaries,
autoScalerEventHandler);
- if (scalingWouldExceedClusterResources(
- configOverrides.newConfigWithOverrides(conf),
+ var memoryTuningEnabled =
conf.get(AutoScalerOptions.MEMORY_TUNING_ENABLED);
+ if (scalingWouldExceedMaxResources(
+ memoryTuningEnabled ?
configOverrides.newConfigWithOverrides(conf) : conf,
+ jobTopology,
evaluatedMetrics,
scalingSummaries,
context)) {
@@ -280,6 +286,30 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
return false;
}
+ @VisibleForTesting
+ protected boolean scalingWouldExceedMaxResources(
+ Configuration tunedConfig,
+ JobTopology jobTopology,
+ EvaluatedMetrics evaluatedMetrics,
+ Map<JobVertexID, ScalingSummary> scalingSummaries,
+ Context ctx) {
+ if (scalingWouldExceedClusterResources(
+ tunedConfig, evaluatedMetrics, scalingSummaries, ctx)) {
+ return true;
+ }
+ if (scalingWouldExceedResourceQuota(tunedConfig, jobTopology,
scalingSummaries, ctx)) {
+ autoScalerEventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Warning,
+ "ResourceQuotaReached",
+ RESOURCE_QUOTA_REACHED_MESSAGE,
+ null,
+ tunedConfig.get(SCALING_EVENT_INTERVAL));
+ return true;
+ }
+ return false;
+ }
+
private boolean scalingWouldExceedClusterResources(
Configuration tunedConfig,
EvaluatedMetrics evaluatedMetrics,
@@ -306,7 +336,7 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
ResourceCheckUtils.estimateNumTaskSlotsAfterRescale(
evaluatedMetrics.getVertexMetrics(), scalingSummaries,
numTaskSlotsUsed);
- int taskSlotsPerTm =
ctx.getConfiguration().get(TaskManagerOptions.NUM_TASK_SLOTS);
+ int taskSlotsPerTm =
tunedConfig.get(TaskManagerOptions.NUM_TASK_SLOTS);
int currentNumTms = (int) Math.ceil(numTaskSlotsUsed / (double)
taskSlotsPerTm);
int newNumTms = (int) Math.ceil(numTaskSlotsAfterRescale / (double)
taskSlotsPerTm);
@@ -315,6 +345,83 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
currentNumTms, newNumTms, taskManagerCpu, taskManagerMemory);
}
+ protected static boolean scalingWouldExceedResourceQuota(
+ Configuration tunedConfig,
+ JobTopology jobTopology,
+ Map<JobVertexID, ScalingSummary> scalingSummaries,
+ JobAutoScalerContext<?> ctx) {
+
+ if (jobTopology == null ||
jobTopology.getSlotSharingGroupMapping().isEmpty()) {
+ return false;
+ }
+
+ var cpuQuota = tunedConfig.getOptional(AutoScalerOptions.CPU_QUOTA);
+ var memoryQuota =
tunedConfig.getOptional(AutoScalerOptions.MEMORY_QUOTA);
+ var tmMemory = MemoryTuning.getTotalMemory(tunedConfig, ctx);
+ var tmCpu = ctx.getTaskManagerCpu().orElse(0.);
+
+ if (cpuQuota.isPresent() || memoryQuota.isPresent()) {
+ var currentSlotSharingGroupMaxParallelisms = new
HashMap<SlotSharingGroupId, Integer>();
+ var newSlotSharingGroupMaxParallelisms = new
HashMap<SlotSharingGroupId, Integer>();
+ for (var e : jobTopology.getSlotSharingGroupMapping().entrySet()) {
+ int currentMaxParallelism =
+ e.getValue().stream()
+ .filter(scalingSummaries::containsKey)
+ .mapToInt(v ->
scalingSummaries.get(v).getCurrentParallelism())
+ .max()
+ .orElse(0);
+ currentSlotSharingGroupMaxParallelisms.put(e.getKey(),
currentMaxParallelism);
+ int newMaxParallelism =
+ e.getValue().stream()
+ .filter(scalingSummaries::containsKey)
+ .mapToInt(v ->
scalingSummaries.get(v).getNewParallelism())
+ .max()
+ .orElse(0);
+ newSlotSharingGroupMaxParallelisms.put(e.getKey(),
newMaxParallelism);
+ }
+
+ var numSlotsPerTm =
tunedConfig.get(TaskManagerOptions.NUM_TASK_SLOTS);
+ var currentTotalSlots =
+ currentSlotSharingGroupMaxParallelisms.values().stream()
+ .mapToInt(Integer::intValue)
+ .sum();
+ var currentNumTms = currentTotalSlots / numSlotsPerTm;
+ var newTotalSlots =
+ newSlotSharingGroupMaxParallelisms.values().stream()
+ .mapToInt(Integer::intValue)
+ .sum();
+ var newNumTms = newTotalSlots / numSlotsPerTm;
+
+ if (newNumTms <= currentNumTms) {
+ LOG.debug(
+ "Skipping quota check due to new resource allocation
is less or equals than the current");
+ return false;
+ }
+
+ if (cpuQuota.isPresent()) {
+ LOG.debug("CPU resource quota is {}, checking limits",
cpuQuota.get());
+ double totalCPU = tmCpu * newNumTms;
+ if (totalCPU > cpuQuota.get()) {
+ LOG.info("CPU resource quota reached with value: {}",
totalCPU);
+ return true;
+ }
+ }
+
+ if (memoryQuota.isPresent()) {
+ LOG.debug("Memory resource quota is {}, checking limits",
memoryQuota.get());
+ long totalMemory = tmMemory.getBytes() * newNumTms;
+ if (totalMemory > memoryQuota.get().getBytes()) {
+ LOG.info(
+ "Memory resource quota reached with value: {}",
+ new MemorySize(totalMemory));
+ return true;
+ }
+ }
+ }
+
+ return false;
+ }
+
private static Map<String, String> getVertexParallelismOverrides(
Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>
evaluatedMetrics,
Map<JobVertexID, ScalingSummary> summaries) {
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
index 0a3fef33..276bf561 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
@@ -210,7 +210,15 @@ public abstract class ScalingMetricCollector<KEY, Context
extends JobAutoScalerC
@VisibleForTesting
@SneakyThrows
protected JobTopology getJobTopology(JobDetailsInfo jobDetailsInfo) {
- Map<JobVertexID, Integer> maxParallelismMap =
+ var slotSharingGroupIdMap =
+ jobDetailsInfo.getJobVertexInfos().stream()
+ .filter(e -> e.getSlotSharingGroupId() != null)
+ .collect(
+ Collectors.toMap(
+
JobDetailsInfo.JobVertexDetailsInfo::getJobVertexID,
+ JobDetailsInfo.JobVertexDetailsInfo
+ ::getSlotSharingGroupId));
+ var maxParallelismMap =
jobDetailsInfo.getJobVertexInfos().stream()
.collect(
Collectors.toMap(
@@ -235,7 +243,8 @@ public abstract class ScalingMetricCollector<KEY, Context
extends JobAutoScalerC
d.getJobVertexID(),
IOMetrics.from(d.getJobVertexMetrics()));
});
- return JobTopology.fromJsonPlan(json, maxParallelismMap, metrics,
finished);
+ return JobTopology.fromJsonPlan(
+ json, slotSharingGroupIdMap, maxParallelismMap, metrics,
finished);
}
private void updateKafkaSourceMaxParallelisms(Context ctx, JobID jobId,
JobTopology topology)
@@ -254,7 +263,7 @@ public abstract class ScalingMetricCollector<KEY, Context
extends JobAutoScalerC
"Updating source {} max parallelism based on
available partitions to {}",
sourceVertex,
numPartitions);
- topology.updateMaxParallelism(sourceVertex, (int)
numPartitions);
+ topology.get(sourceVertex).updateMaxParallelism((int)
numPartitions);
}
}
}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
index aa095e1c..6922448b 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
@@ -20,6 +20,7 @@ package org.apache.flink.autoscaler.config;
import org.apache.flink.autoscaler.metrics.MetricAggregator;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.MemorySize;
import java.time.Duration;
import java.util.List;
@@ -327,4 +328,20 @@ public class AutoScalerOptions {
.defaultValue(Duration.ofSeconds(10))
.withFallbackKeys(oldOperatorConfigKey("flink.rest-client.timeout"))
.withDescription("The timeout for waiting the flink rest
client to return.");
+
+ public static final ConfigOption<MemorySize> MEMORY_QUOTA =
+ autoScalerConfig("quota.memory")
+ .memoryType()
+ .noDefaultValue()
+ .withFallbackKeys(oldOperatorConfigKey("quota.memory"))
+ .withDescription(
+ "Quota of the memory size. When scaling would go
beyond this number the the scaling is not going to happen.");
+
+ public static final ConfigOption<Double> CPU_QUOTA =
+ autoScalerConfig("quota.cpu")
+ .doubleType()
+ .noDefaultValue()
+ .withFallbackKeys(oldOperatorConfigKey("quota.cpu"))
+ .withDescription(
+ "Quota of the CPU count. When scaling would go
beyond this number the the scaling is not going to happen.");
}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java
index 8945882c..851db6b0 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java
@@ -17,6 +17,7 @@
package org.apache.flink.autoscaler.topology;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
@@ -48,6 +49,7 @@ public class JobTopology {
private static final ObjectMapper objectMapper = new ObjectMapper();
@Getter private final Map<JobVertexID, VertexInfo> vertexInfos;
+ @Getter private final Map<SlotSharingGroupId, Set<JobVertexID>>
slotSharingGroupMapping;
@Getter private final Set<JobVertexID> finishedVertices;
@Getter private final List<JobVertexID> verticesInTopologicalOrder;
@@ -66,6 +68,7 @@ public class JobTopology {
ImmutableMap.copyOf(
vertexInfo.stream().collect(Collectors.toMap(VertexInfo::getId, v -> v)));
+ Map<SlotSharingGroupId, Set<JobVertexID>>
vertexSlotSharingGroupMapping = new HashMap<>();
var finishedVertices = ImmutableSet.<JobVertexID>builder();
vertexInfo.forEach(
@@ -79,12 +82,21 @@ public class JobTopology {
vertexOutputs
.computeIfAbsent(inputId,
id -> new HashMap<>())
.put(vertexId,
shipStrategy));
+
+ var slotSharingGroupId = info.getSlotSharingGroupId();
+ if (slotSharingGroupId != null) {
+ vertexSlotSharingGroupMapping
+ .computeIfAbsent(slotSharingGroupId, id -> new
HashSet<>())
+ .add(vertexId);
+ }
+
if (info.isFinished()) {
finishedVertices.add(vertexId);
}
});
vertexOutputs.forEach((v, outputs) ->
vertexInfos.get(v).setOutputs(outputs));
+ this.slotSharingGroupMapping =
ImmutableMap.copyOf(vertexSlotSharingGroupMapping);
this.finishedVertices = finishedVertices.build();
this.verticesInTopologicalOrder = returnVerticesInTopologicalOrder();
}
@@ -97,10 +109,6 @@ public class JobTopology {
return get(jobVertexID).getInputs().isEmpty();
}
- public void updateMaxParallelism(JobVertexID vertexID, int maxParallelism)
{
- get(vertexID).updateMaxParallelism(maxParallelism);
- }
-
private List<JobVertexID> returnVerticesInTopologicalOrder() {
List<JobVertexID> sorted = new ArrayList<>(vertexInfos.size());
@@ -134,6 +142,7 @@ public class JobTopology {
public static JobTopology fromJsonPlan(
String jsonPlan,
+ Map<JobVertexID, SlotSharingGroupId> slotSharingGroupIdMap,
Map<JobVertexID, Integer> maxParallelismMap,
Map<JobVertexID, IOMetrics> metrics,
Set<JobVertexID> finishedVertices)
@@ -151,6 +160,7 @@ public class JobTopology {
vertexInfo.add(
new VertexInfo(
vertexId,
+ slotSharingGroupIdMap.get(vertexId),
inputs,
node.get("parallelism").asInt(),
maxParallelismMap.get(vertexId),
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java
index 2428cdeb..705dd4c4 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java
@@ -18,9 +18,12 @@
package org.apache.flink.autoscaler.topology;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import lombok.AccessLevel;
import lombok.Data;
+import lombok.Setter;
import java.util.Map;
@@ -33,11 +36,14 @@ public class VertexInfo {
// All input vertices and the ship_strategy
private final Map<JobVertexID, ShipStrategy> inputs;
+ private final SlotSharingGroupId slotSharingGroupId;
+
// All output vertices and the ship_strategy
private Map<JobVertexID, ShipStrategy> outputs;
private final int parallelism;
+ @Setter(AccessLevel.NONE)
private int maxParallelism;
private final int originalMaxParallelism;
@@ -48,12 +54,14 @@ public class VertexInfo {
public VertexInfo(
JobVertexID id,
+ SlotSharingGroupId slotSharingGroupId,
Map<JobVertexID, ShipStrategy> inputs,
int parallelism,
int maxParallelism,
boolean finished,
IOMetrics ioMetrics) {
this.id = id;
+ this.slotSharingGroupId = slotSharingGroupId;
this.inputs = inputs;
this.parallelism = parallelism;
this.maxParallelism = maxParallelism;
@@ -69,7 +77,18 @@ public class VertexInfo {
int parallelism,
int maxParallelism,
IOMetrics ioMetrics) {
- this(id, inputs, parallelism, maxParallelism, false, ioMetrics);
+ this(id, null, inputs, parallelism, maxParallelism, false, ioMetrics);
+ }
+
+ @VisibleForTesting
+ public VertexInfo(
+ JobVertexID id,
+ Map<JobVertexID, ShipStrategy> inputs,
+ int parallelism,
+ int maxParallelism,
+ boolean finished,
+ IOMetrics ioMetrics) {
+ this(id, null, inputs, parallelism, maxParallelism, finished,
ioMetrics);
}
@VisibleForTesting
@@ -82,6 +101,6 @@ public class VertexInfo {
}
public void updateMaxParallelism(int maxParallelism) {
- setMaxParallelism(Math.min(originalMaxParallelism, maxParallelism));
+ this.maxParallelism = Math.min(originalMaxParallelism, maxParallelism);
}
}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
new file mode 100644
index 00000000..c6a16e67
--- /dev/null
+++
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
@@ -0,0 +1,448 @@
+/*
+ * 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;
+
+// import io.swagger.v3.oas.annotations.media.Schema;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.job.metrics.IOMetricsInfo;
+import org.apache.flink.runtime.rest.messages.json.JobIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.JobIDSerializer;
+import org.apache.flink.runtime.rest.messages.json.JobVertexIDDeserializer;
+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.SlotSharingGroupIDSerializer;
+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 java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+
+/** Copied from Flink. Should be removed once the client dependency is
upgraded to 1.18. */
+/** The difference compared to 1.18 is that slot sharing group is optional
here. */
+public class JobDetailsInfo implements ResponseBody {
+
+ public static final String FIELD_NAME_JOB_ID = "jid";
+
+ public static final String FIELD_NAME_JOB_NAME = "name";
+
+ public static final String FIELD_NAME_IS_STOPPABLE = "isStoppable";
+
+ public static final String FIELD_NAME_JOB_STATUS = "state";
+
+ public static final String FIELD_NAME_START_TIME = "start-time";
+
+ public static final String FIELD_NAME_END_TIME = "end-time";
+
+ public static final String FIELD_NAME_DURATION = "duration";
+
+ public static final String FIELD_NAME_MAX_PARALLELISM = "maxParallelism";
+
+ // TODO: For what do we need this???
+ public static final String FIELD_NAME_NOW = "now";
+
+ public static final String FIELD_NAME_TIMESTAMPS = "timestamps";
+
+ public static final String FIELD_NAME_JOB_VERTEX_INFOS = "vertices";
+
+ public static final String FIELD_NAME_JOB_VERTICES_PER_STATE =
"status-counts";
+
+ public static final String FIELD_NAME_JSON_PLAN = "plan";
+
+ @JsonProperty(FIELD_NAME_JOB_ID)
+ @JsonSerialize(using = JobIDSerializer.class)
+ private final JobID jobId;
+
+ @JsonProperty(FIELD_NAME_JOB_NAME)
+ private final String name;
+
+ @JsonProperty(FIELD_NAME_IS_STOPPABLE)
+ private final boolean isStoppable;
+
+ @JsonProperty(FIELD_NAME_JOB_STATUS)
+ private final JobStatus jobStatus;
+
+ @JsonProperty(FIELD_NAME_START_TIME)
+ private final long startTime;
+
+ @JsonProperty(FIELD_NAME_END_TIME)
+ private final long endTime;
+
+ @JsonProperty(FIELD_NAME_DURATION)
+ private final long duration;
+
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM)
+ private final long maxParallelism;
+
+ @JsonProperty(FIELD_NAME_NOW)
+ private final long now;
+
+ @JsonProperty(FIELD_NAME_TIMESTAMPS)
+ private final Map<JobStatus, Long> timestamps;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_INFOS)
+ private final Collection<JobVertexDetailsInfo> jobVertexInfos;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTICES_PER_STATE)
+ private final Map<ExecutionState, Integer> jobVerticesPerState;
+
+ @JsonProperty(FIELD_NAME_JSON_PLAN)
+ private final JobPlanInfo.RawJson jsonPlan;
+
+ @JsonCreator
+ public JobDetailsInfo(
+ @JsonDeserialize(using = JobIDDeserializer.class)
@JsonProperty(FIELD_NAME_JOB_ID)
+ JobID jobId,
+ @JsonProperty(FIELD_NAME_JOB_NAME) String name,
+ @JsonProperty(FIELD_NAME_IS_STOPPABLE) boolean isStoppable,
+ @JsonProperty(FIELD_NAME_JOB_STATUS) JobStatus jobStatus,
+ @JsonProperty(FIELD_NAME_START_TIME) long startTime,
+ @JsonProperty(FIELD_NAME_END_TIME) long endTime,
+ @JsonProperty(FIELD_NAME_DURATION) long duration,
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM) long maxParallelism,
+ @JsonProperty(FIELD_NAME_NOW) long now,
+ @JsonProperty(FIELD_NAME_TIMESTAMPS) Map<JobStatus, Long>
timestamps,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_INFOS)
+ Collection<JobVertexDetailsInfo> jobVertexInfos,
+ @JsonProperty(FIELD_NAME_JOB_VERTICES_PER_STATE)
+ Map<ExecutionState, Integer> jobVerticesPerState,
+ @JsonProperty(FIELD_NAME_JSON_PLAN) JobPlanInfo.RawJson jsonPlan) {
+ this.jobId = Preconditions.checkNotNull(jobId);
+ this.name = Preconditions.checkNotNull(name);
+ this.isStoppable = isStoppable;
+ this.jobStatus = Preconditions.checkNotNull(jobStatus);
+ this.startTime = startTime;
+ this.endTime = endTime;
+ this.duration = duration;
+ this.maxParallelism = maxParallelism;
+ this.now = now;
+ this.timestamps = Preconditions.checkNotNull(timestamps);
+ this.jobVertexInfos = Preconditions.checkNotNull(jobVertexInfos);
+ this.jobVerticesPerState =
Preconditions.checkNotNull(jobVerticesPerState);
+ this.jsonPlan = Preconditions.checkNotNull(jsonPlan);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ JobDetailsInfo that = (JobDetailsInfo) o;
+ return isStoppable == that.isStoppable
+ && startTime == that.startTime
+ && endTime == that.endTime
+ && duration == that.duration
+ && maxParallelism == that.maxParallelism
+ && now == that.now
+ && Objects.equals(jobId, that.jobId)
+ && Objects.equals(name, that.name)
+ && jobStatus == that.jobStatus
+ && Objects.equals(timestamps, that.timestamps)
+ && Objects.equals(jobVertexInfos, that.jobVertexInfos)
+ && Objects.equals(jobVerticesPerState,
that.jobVerticesPerState)
+ && Objects.equals(jsonPlan, that.jsonPlan);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ jobId,
+ name,
+ isStoppable,
+ jobStatus,
+ startTime,
+ endTime,
+ duration,
+ maxParallelism,
+ now,
+ timestamps,
+ jobVertexInfos,
+ jobVerticesPerState,
+ jsonPlan);
+ }
+
+ @JsonIgnore
+ public JobID getJobId() {
+ return jobId;
+ }
+
+ @JsonIgnore
+ public String getName() {
+ return name;
+ }
+
+ @JsonIgnore
+ public boolean isStoppable() {
+ return isStoppable;
+ }
+
+ @JsonIgnore
+ public JobStatus getJobStatus() {
+ return jobStatus;
+ }
+
+ @JsonIgnore
+ public long getStartTime() {
+ return startTime;
+ }
+
+ @JsonIgnore
+ public long getEndTime() {
+ return endTime;
+ }
+
+ @JsonIgnore
+ public long getMaxParallelism() {
+ return maxParallelism;
+ }
+
+ @JsonIgnore
+ public long getDuration() {
+ return duration;
+ }
+
+ @JsonIgnore
+ public long getNow() {
+ return now;
+ }
+
+ @JsonIgnore
+ public Map<JobStatus, Long> getTimestamps() {
+ return timestamps;
+ }
+
+ @JsonIgnore
+ public Collection<JobVertexDetailsInfo> getJobVertexInfos() {
+ return jobVertexInfos;
+ }
+
+ @JsonIgnore
+ public Map<ExecutionState, Integer> getJobVerticesPerState() {
+ return jobVerticesPerState;
+ }
+
+ @JsonIgnore
+ public String getJsonPlan() {
+ return jsonPlan.toString();
+ }
+
+ // ---------------------------------------------------
+ // Static inner classes
+ // ---------------------------------------------------
+
+ /** Detailed information about a job vertex. */
+ // @Schema(name = "JobDetailsVertexInfo")
+ public static final class JobVertexDetailsInfo {
+
+ public static final String FIELD_NAME_JOB_VERTEX_ID = "id";
+
+ public static final String FIELD_NAME_SLOT_SHARING_GROUP_ID =
"slotSharingGroupId";
+
+ public static final String FIELD_NAME_JOB_VERTEX_NAME = "name";
+
+ public static final String FIELD_NAME_MAX_PARALLELISM =
"maxParallelism";
+
+ public static final String FIELD_NAME_PARALLELISM = "parallelism";
+
+ public static final String FIELD_NAME_JOB_VERTEX_STATE = "status";
+
+ public static final String FIELD_NAME_JOB_VERTEX_START_TIME =
"start-time";
+
+ public static final String FIELD_NAME_JOB_VERTEX_END_TIME = "end-time";
+
+ public static final String FIELD_NAME_JOB_VERTEX_DURATION = "duration";
+
+ public static final String FIELD_NAME_TASKS_PER_STATE = "tasks";
+
+ public static final String FIELD_NAME_JOB_VERTEX_METRICS = "metrics";
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_ID)
+ @JsonSerialize(using = JobVertexIDSerializer.class)
+ private final JobVertexID jobVertexID;
+
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_ID)
+ @JsonSerialize(using = SlotSharingGroupIDSerializer.class)
+ private final SlotSharingGroupId slotSharingGroupId;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_NAME)
+ private final String name;
+
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM)
+ private final int maxParallelism;
+
+ @JsonProperty(FIELD_NAME_PARALLELISM)
+ private final int parallelism;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_STATE)
+ private final ExecutionState executionState;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_START_TIME)
+ private final long startTime;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_END_TIME)
+ private final long endTime;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_DURATION)
+ private final long duration;
+
+ @JsonProperty(FIELD_NAME_TASKS_PER_STATE)
+ private final Map<ExecutionState, Integer> tasksPerState;
+
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_METRICS)
+ private final IOMetricsInfo jobVertexMetrics;
+
+ @JsonCreator
+ public JobVertexDetailsInfo(
+ @JsonDeserialize(using = JobVertexIDDeserializer.class)
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_ID)
+ JobVertexID jobVertexID,
+ @JsonDeserialize(using = SlotSharingGroupIDDeserializer.class)
+ @JsonProperty(FIELD_NAME_SLOT_SHARING_GROUP_ID)
+ SlotSharingGroupId slotSharingGroupId,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_NAME) String name,
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM) int maxParallelism,
+ @JsonProperty(FIELD_NAME_PARALLELISM) int parallelism,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_STATE) ExecutionState
executionState,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_START_TIME) long startTime,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_END_TIME) long endTime,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_DURATION) long duration,
+ @JsonProperty(FIELD_NAME_TASKS_PER_STATE)
+ Map<ExecutionState, Integer> tasksPerState,
+ @JsonProperty(FIELD_NAME_JOB_VERTEX_METRICS) IOMetricsInfo
jobVertexMetrics) {
+ this.jobVertexID = Preconditions.checkNotNull(jobVertexID);
+ this.slotSharingGroupId = slotSharingGroupId;
+ this.name = Preconditions.checkNotNull(name);
+ this.maxParallelism = maxParallelism;
+ this.parallelism = parallelism;
+ this.executionState = Preconditions.checkNotNull(executionState);
+ this.startTime = startTime;
+ this.endTime = endTime;
+ this.duration = duration;
+ this.tasksPerState = Preconditions.checkNotNull(tasksPerState);
+ this.jobVertexMetrics =
Preconditions.checkNotNull(jobVertexMetrics);
+ }
+
+ @JsonIgnore
+ public JobVertexID getJobVertexID() {
+ return jobVertexID;
+ }
+
+ @JsonIgnore
+ public SlotSharingGroupId getSlotSharingGroupId() {
+ return slotSharingGroupId;
+ }
+
+ @JsonIgnore
+ public String getName() {
+ return name;
+ }
+
+ @JsonIgnore
+ public int getMaxParallelism() {
+ return maxParallelism;
+ }
+
+ @JsonIgnore
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ @JsonIgnore
+ public ExecutionState getExecutionState() {
+ return executionState;
+ }
+
+ @JsonIgnore
+ public long getStartTime() {
+ return startTime;
+ }
+
+ @JsonIgnore
+ public long getEndTime() {
+ return endTime;
+ }
+
+ @JsonIgnore
+ public long getDuration() {
+ return duration;
+ }
+
+ @JsonIgnore
+ public Map<ExecutionState, Integer> getTasksPerState() {
+ return tasksPerState;
+ }
+
+ @JsonIgnore
+ public IOMetricsInfo getJobVertexMetrics() {
+ return jobVertexMetrics;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ JobVertexDetailsInfo that = (JobVertexDetailsInfo) o;
+ return maxParallelism == that.maxParallelism
+ && parallelism == that.parallelism
+ && startTime == that.startTime
+ && endTime == that.endTime
+ && duration == that.duration
+ && Objects.equals(jobVertexID, that.jobVertexID)
+ && Objects.equals(slotSharingGroupId,
that.slotSharingGroupId)
+ && Objects.equals(name, that.name)
+ && executionState == that.executionState
+ && Objects.equals(tasksPerState, that.tasksPerState)
+ && Objects.equals(jobVertexMetrics, that.jobVertexMetrics);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ jobVertexID,
+ slotSharingGroupId,
+ name,
+ maxParallelism,
+ parallelism,
+ executionState,
+ startTime,
+ endTime,
+ duration,
+ tasksPerState,
+ jobVertexMetrics);
+ }
+ }
+}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDDeserializer.java
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDDeserializer.java
new file mode 100644
index 00000000..c2d40a5c
--- /dev/null
+++
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDDeserializer.java
@@ -0,0 +1,44 @@
+/*
+ * 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.json;
+
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+
+import java.io.IOException;
+
+/** Copied from Flink. Should be removed once the client dependency is
upgraded to 1.18. */
+public class SlotSharingGroupIDDeserializer extends
StdDeserializer<SlotSharingGroupId> {
+
+ private static final long serialVersionUID = -2908308366715321301L;
+
+ protected SlotSharingGroupIDDeserializer() {
+ super(JobVertexID.class);
+ }
+
+ @Override
+ public SlotSharingGroupId deserialize(JsonParser p, DeserializationContext
ctxt)
+ throws IOException {
+ return SlotSharingGroupIdConverter.fromHexString(p.getValueAsString());
+ }
+}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDSerializer.java
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDSerializer.java
new file mode 100644
index 00000000..4d1f51d1
--- /dev/null
+++
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIDSerializer.java
@@ -0,0 +1,43 @@
+/*
+ * 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.json;
+
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+
+/** Copied from Flink. Should be removed once the client dependency is
upgraded to 1.18. */
+public class SlotSharingGroupIDSerializer extends
StdSerializer<SlotSharingGroupId> {
+
+ private static final long serialVersionUID = -4052148694985726120L;
+
+ public SlotSharingGroupIDSerializer() {
+ super(SlotSharingGroupId.class);
+ }
+
+ @Override
+ public void serialize(SlotSharingGroupId value, JsonGenerator gen,
SerializerProvider provider)
+ throws IOException {
+ gen.writeString(value.toString());
+ }
+}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIdConverter.java
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIdConverter.java
new file mode 100644
index 00000000..38d9fbf9
--- /dev/null
+++
b/flink-autoscaler/src/main/java/org/apache/flink/runtime/rest/messages/json/SlotSharingGroupIdConverter.java
@@ -0,0 +1,42 @@
+/*
+ * 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.json;
+
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
+import org.apache.flink.util.StringUtils;
+
+/** Copied from Flink. Should be removed once the client dependency is
upgraded to 1.18. */
+public class SlotSharingGroupIdConverter {
+ public static SlotSharingGroupId fromHexString(String hexString) {
+ var bytes = StringUtils.hexStringToByte(hexString);
+ var lowerPart = byteArrayToLong(bytes, 0);
+ var upperPart = byteArrayToLong(bytes, 8);
+
+ return new SlotSharingGroupId(lowerPart, upperPart);
+ }
+
+ private static long byteArrayToLong(byte[] ba, int offset) {
+ long l = 0L;
+
+ for (int i = 0; i < 8; ++i) {
+ l |= ((long) ba[offset + 8 - 1 - i] & 255L) << (i << 3);
+ }
+
+ return l;
+ }
+}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
index 97631b05..df2b1049 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -31,6 +31,7 @@ import
org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.autoscaler.topology.IOMetrics;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.autoscaler.topology.VertexInfo;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
@@ -68,6 +69,7 @@ public class MetricsCollectionAndEvaluationTest {
private InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>
stateStore;
private JobVertexID source1, source2, map, sink;
+ private SlotSharingGroupId slotSharingGroupId;
private JobTopology topology;
private Clock clock;
@@ -86,6 +88,7 @@ public class MetricsCollectionAndEvaluationTest {
source2 = new JobVertexID();
map = new JobVertexID();
sink = new JobVertexID();
+ slotSharingGroupId = new SlotSharingGroupId();
topology =
new JobTopology(
@@ -415,7 +418,13 @@ public class MetricsCollectionAndEvaluationTest {
new JobTopology(
new VertexInfo(s1, Map.of(), 10, 720),
new VertexInfo(
- finished, Map.of(), 10, 720, true,
IOMetrics.FINISHED_METRICS));
+ finished,
+ null,
+ Map.of(),
+ 10,
+ 720,
+ true,
+ IOMetrics.FINISHED_METRICS));
metricsCollector = new TestingMetricsCollector(topology);
metricsCollector.setJobUpdateTs(startTime);
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index 26a92f3e..97972269 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -32,11 +32,14 @@ import org.apache.flink.autoscaler.topology.VertexInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
@@ -46,7 +49,9 @@ import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static
org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static
org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_REPORT_REASON;
@@ -58,7 +63,9 @@ import static
org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
/** Test for {@link ScalingExecutor}. */
public class ScalingExecutorTest {
@@ -73,6 +80,8 @@ public class ScalingExecutorTest {
private Configuration conf;
+ private Configuration capturedConfForMaxResources;
+
private ScalingTracking scalingTracking = new ScalingTracking();
private static final Map<ScalingMetric, EvaluatedScalingMetric>
dummyGlobalMetrics =
@@ -86,7 +95,21 @@ public class ScalingExecutorTest {
context = createDefaultJobAutoScalerContext();
stateStore = new InMemoryAutoScalerStateStore<>();
- scalingExecutor = new ScalingExecutor<>(eventCollector, stateStore);
+ capturedConfForMaxResources = null;
+ scalingExecutor =
+ new ScalingExecutor<>(eventCollector, stateStore) {
+ @Override
+ protected boolean scalingWouldExceedMaxResources(
+ Configuration tunedConfig,
+ JobTopology jobTopology,
+ EvaluatedMetrics evaluatedMetrics,
+ Map<JobVertexID, ScalingSummary> scalingSummaries,
+ JobAutoScalerContext<JobID> ctx) {
+ capturedConfForMaxResources = tunedConfig;
+ return super.scalingWouldExceedMaxResources(
+ tunedConfig, jobTopology, evaluatedMetrics,
scalingSummaries, ctx);
+ }
+ };
conf = context.getConfiguration();
conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
conf.set(AutoScalerOptions.SCALING_ENABLED, true);
@@ -262,6 +285,7 @@ public class ScalingExecutorTest {
var conf = context.getConfiguration();
var now = Instant.now();
var localTime = ZonedDateTime.ofInstant(now,
ZoneId.systemDefault()).toLocalTime();
+
// scaling execution in excluded periods
var excludedPeriod =
new StringBuilder(localTime.toString().split("\\.")[0])
@@ -358,10 +382,12 @@ public class ScalingExecutorTest {
jobTopology));
}
- @Test
- public void testMemoryTuning() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testMemoryTuning(boolean memoryTuningEnabled) throws Exception
{
context = TestingAutoscalerUtils.createResourceAwareContext();
-
context.getConfiguration().set(AutoScalerOptions.MEMORY_TUNING_ENABLED, true);
+ context.getConfiguration()
+ .set(AutoScalerOptions.MEMORY_TUNING_ENABLED,
memoryTuningEnabled);
context.getConfiguration().set(TaskManagerOptions.NUM_TASK_SLOTS, 5);
context.getConfiguration()
.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY,
MemorySize.parse("30 gb"));
@@ -386,13 +412,13 @@ public class ScalingExecutorTest {
EvaluatedScalingMetric.of(Double.NaN));
var vertexMetrics =
Map.of(source, evaluated(10, 100, 50, 0), sink, evaluated(10,
100, 50, 0));
- var metrics = new EvaluatedMetrics(vertexMetrics, globalMetrics);
- JobTopology jobTopology =
+ var jobTopology =
new JobTopology(
new VertexInfo(source, Map.of(), 10, 1000, false,
null),
new VertexInfo(sink, Map.of(source, REBALANCE), 10,
1000, false, null));
+ var metrics = new EvaluatedMetrics(vertexMetrics, globalMetrics);
assertTrue(
scalingExecutor.scaleResource(
context,
@@ -401,23 +427,31 @@ public class ScalingExecutorTest {
new ScalingTracking(),
now,
jobTopology));
+ Map<String, String> expected;
+ if (memoryTuningEnabled) {
+ assertNotEquals(context.getConfiguration(),
capturedConfForMaxResources);
+ expected =
+ Map.of(
+ TaskManagerOptions.MANAGED_MEMORY_FRACTION.key(),
+ "0.652",
+ TaskManagerOptions.NETWORK_MEMORY_MIN.key(),
+ "24320 kb",
+ TaskManagerOptions.NETWORK_MEMORY_MAX.key(),
+ "24320 kb",
+ TaskManagerOptions.JVM_METASPACE.key(),
+ "360 mb",
+ TaskManagerOptions.JVM_OVERHEAD_FRACTION.key(),
+ "0.053",
+ TaskManagerOptions.FRAMEWORK_HEAP_MEMORY.key(),
+ "0 bytes",
+ TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
+ "20400832696 bytes");
+ } else {
+ assertEquals(context.getConfiguration(),
capturedConfForMaxResources);
+ expected = Map.of();
+ }
assertThat(stateStore.getConfigChanges(context).getOverrides())
- .containsExactlyInAnyOrderEntriesOf(
- Map.of(
-
TaskManagerOptions.MANAGED_MEMORY_FRACTION.key(),
- "0.652",
- TaskManagerOptions.NETWORK_MEMORY_MIN.key(),
- "24320 kb",
- TaskManagerOptions.NETWORK_MEMORY_MAX.key(),
- "24320 kb",
- TaskManagerOptions.JVM_METASPACE.key(),
- "360 mb",
- TaskManagerOptions.JVM_OVERHEAD_FRACTION.key(),
- "0.053",
- TaskManagerOptions.FRAMEWORK_HEAP_MEMORY.key(),
- "0 bytes",
- TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
- "20400832696 bytes"));
+ .containsExactlyInAnyOrderEntriesOf(expected);
}
@ParameterizedTest
@@ -675,6 +709,121 @@ public class ScalingExecutorTest {
"8"));
}
+ @ParameterizedTest
+ @MethodSource("testDataForQuota")
+ public void testQuota(
+ SlotSharingGroupId slotSharingGroupId1,
+ SlotSharingGroupId slotSharingGroupId2,
+ Optional<Double> cpuQuota,
+ Optional<String> memoryQuota,
+ boolean quotaReached)
+ throws Exception {
+
+ var ctx = TestingAutoscalerUtils.createResourceAwareContext(2., "2g");
+ var conf = ctx.getConfiguration();
+ conf.setString("taskmanager.numberOfTaskSlots", "2");
+ cpuQuota.ifPresent(v -> conf.set(AutoScalerOptions.CPU_QUOTA, v));
+ memoryQuota.ifPresent(v -> conf.set(AutoScalerOptions.MEMORY_QUOTA,
MemorySize.parse(v)));
+ conf.set(AutoScalerOptions.TARGET_UTILIZATION, 0.6);
+ conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.);
+
+ testQuotaReached(slotSharingGroupId1, slotSharingGroupId2,
quotaReached, ctx);
+ }
+
+ private static Stream<Arguments> testDataForQuota() {
+ var slotSharingGroupId1 = new SlotSharingGroupId();
+ var slotSharingGroupId2 = new SlotSharingGroupId();
+ return Stream.of(
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId1,
+ Optional.of(30.),
+ Optional.empty(),
+ false),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId2,
+ Optional.of(50.),
+ Optional.empty(),
+ false),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId1,
+ Optional.empty(),
+ Optional.of("30g"),
+ false),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId2,
+ Optional.empty(),
+ Optional.of("50g"),
+ false),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId1,
+ Optional.of(3.),
+ Optional.empty(),
+ true),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId2,
+ Optional.of(5.),
+ Optional.empty(),
+ true),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId1,
+ Optional.empty(),
+ Optional.of("3g"),
+ true),
+ arguments(
+ slotSharingGroupId1,
+ slotSharingGroupId2,
+ Optional.empty(),
+ Optional.of("5g"),
+ true));
+ }
+
+ private void testQuotaReached(
+ SlotSharingGroupId slotSharingGroupId1,
+ SlotSharingGroupId slotSharingGroupId2,
+ boolean quotaReached,
+ JobAutoScalerContext<JobID> ctx)
+ throws Exception {
+ var op1 = new JobVertexID();
+ var op2 = new JobVertexID();
+ var jobTopology =
+ new JobTopology(
+ new VertexInfo(op1, slotSharingGroupId1, Map.of(), 1,
720, false, null),
+ new VertexInfo(op2, slotSharingGroupId2, Map.of(), 1,
720, false, null));
+ var vertexMetrics = Map.of(op1, evaluated(1, 210, 100), op2,
evaluated(1, 110, 100));
+ var metrics =
+ new EvaluatedMetrics(
+ vertexMetrics,
+ Map.of(
+ ScalingMetric.NUM_TASK_SLOTS_USED,
+ EvaluatedScalingMetric.of(0.),
+ ScalingMetric.GC_PRESSURE,
+ EvaluatedScalingMetric.of(Double.NaN),
+ ScalingMetric.HEAP_MAX_USAGE_RATIO,
+ EvaluatedScalingMetric.of(Double.NaN)));
+
+ assertEquals(
+ !quotaReached,
+ scalingExecutor.scaleResource(
+ ctx,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology));
+ if (quotaReached) {
+ assertEquals("ScalingReport",
eventCollector.events.poll().getReason());
+ assertEquals("ResourceQuotaReached",
eventCollector.events.poll().getReason());
+ assertTrue(eventCollector.events.isEmpty());
+ }
+ }
+
private Map<ScalingMetric, EvaluatedScalingMetric> evaluated(
int parallelism, double target, double trueProcessingRate, double
catchupRate) {
var metrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
index 7bc1190c..d5a05b40 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
@@ -228,6 +228,16 @@ public class ScalingMetricCollectorTest {
new JobTopology(source, sink),
metricsCollector.getJobTopology(jobDetailsInfo));
}
+ @Test
+ public void testJobTopologyParsingFromJobDetailsWithSlotSharingGroup()
throws Exception {
+ String s =
+
"{\"jid\":\"a1b1b53c7c71e7199aa8c43bc703fe7f\",\"name\":\"basic-example\",\"isStoppable\":false,\"state\":\"RUNNING\",\"start-time\":1697114719143,\"end-time\":-1,\"duration\":60731,\"maxParallelism\":-1,\"now\":1697114779874,\"timestamps\":{\"CANCELLING\":0,\"INITIALIZING\":1697114719143,\"RUNNING\":1697114719743,\"CANCELED\":0,\"FINISHED\":0,\"FAILED\":0,\"RESTARTING\":0,\"FAILING\":0,\"CREATED\":1697114719343,\"SUSPENDED\":0,\"RECONCILING\":0},\"vertices\":[{\"id\":\"b
[...]
+ JobDetailsInfo jobDetailsInfo = new ObjectMapper().readValue(s,
JobDetailsInfo.class);
+
+ var metricsCollector = new RestApiMetricsCollector();
+ metricsCollector.getJobTopology(jobDetailsInfo);
+ }
+
@Test
public void testJobUpdateTsLogic() {
var details =
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingAutoscalerUtils.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingAutoscalerUtils.java
index 9869e4a3..d1b18b41 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingAutoscalerUtils.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingAutoscalerUtils.java
@@ -52,6 +52,11 @@ public class TestingAutoscalerUtils {
}
public static JobAutoScalerContext<JobID> createResourceAwareContext() {
+ return createResourceAwareContext(100., "30 gb");
+ }
+
+ public static JobAutoScalerContext<JobID> createResourceAwareContext(
+ Double cpu, String memSize) {
JobID jobId = JobID.generate();
MetricRegistry registry = NoOpMetricRegistry.INSTANCE;
GenericMetricGroup metricGroup = new GenericMetricGroup(registry,
null, "test");
@@ -64,12 +69,12 @@ public class TestingAutoscalerUtils {
TestingAutoscalerUtils.getRestClusterClientSupplier()) {
@Override
public Optional<Double> getTaskManagerCpu() {
- return Optional.of(100.);
+ return Optional.ofNullable(cpu);
}
@Override
public Optional<MemorySize> getTaskManagerMemory() {
- return Optional.of(MemorySize.parse("30 gb"));
+ return Optional.ofNullable(memSize).map(MemorySize::parse);
}
};
}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/topology/JobTopologyTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/topology/JobTopologyTest.java
index f857361a..19fa8445 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/topology/JobTopologyTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/topology/JobTopologyTest.java
@@ -79,7 +79,7 @@ public class JobTopologyTest {
JobTopology jobTopology =
JobTopology.fromJsonPlan(
- jsonPlan, maxParallelism, Map.of(),
Collections.emptySet());
+ jsonPlan, Map.of(), maxParallelism, Map.of(),
Collections.emptySet());
assertTrue(jobTopology.get(vertices.get("Sink:
sink1")).getOutputs().isEmpty());
assertTrue(jobTopology.get(vertices.get("Sink:
sink2")).getOutputs().isEmpty());
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
index 835b2200..a1206b8c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
@@ -469,21 +469,23 @@ public class FlinkConfigBuilder {
}
boolean newConfKeys =
spec.getFlinkVersion().isEqualOrNewer(FlinkVersion.v1_17);
- String configKey;
+ String configKey = null;
if (isJM) {
- if (newConfKeys) {
- configKey = KubernetesConfigOptions.JOB_MANAGER_CPU.key();
- } else {
+ // Set new config all the time to simplify reading side
+ conf.setDouble(KubernetesConfigOptions.JOB_MANAGER_CPU.key(),
resource.getCpu());
+ if (!newConfKeys) {
configKey = "kubernetes.jobmanager.cpu";
}
} else {
- if (newConfKeys) {
- configKey = KubernetesConfigOptions.TASK_MANAGER_CPU.key();
- } else {
+ // Set new config all the time to simplify reading side
+ conf.setDouble(KubernetesConfigOptions.TASK_MANAGER_CPU.key(),
resource.getCpu());
+ if (!newConfKeys) {
configKey = "kubernetes.taskmanager.cpu";
}
}
- conf.setDouble(configKey, resource.getCpu());
+ if (configKey != null) {
+ conf.setDouble(configKey, resource.getCpu());
+ }
}
@VisibleForTesting
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
index a723ca93..88a7b4c0 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
@@ -249,6 +249,7 @@ public class EventRecorder {
ScalingReport,
IneffectiveScaling,
MemoryPressure,
+ ResourceQuotaReached,
AutoscalerError,
Scaling,
UnsupportedFlinkVersion
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
index 04a7c842..4f7d9351 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
@@ -421,6 +421,10 @@ public class FlinkConfigBuilderTest {
assertEquals("1.0", confMap.get("kubernetes.jobmanager.cpu"));
assertEquals("1.0", confMap.get("kubernetes.taskmanager.cpu"));
+ // Set new config all the time to simplify reading side
+ assertEquals(Double.valueOf(1),
configuration.get(KubernetesConfigOptions.JOB_MANAGER_CPU));
+ assertEquals(
+ Double.valueOf(1),
configuration.get(KubernetesConfigOptions.TASK_MANAGER_CPU));
}
@Test
