You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2024/02/23 15:45:28 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-34502][autoscaler] Support calculating network memory for forward and rescale edge (#782)
This is an automated email from the ASF dual-hosted git repository.
fanrui 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 304fca82 [FLINK-34502][autoscaler] Support calculating network memory for forward and rescale edge (#782)
304fca82 is described below
commit 304fca82ccc153c0745c468893a8930e9b9be806
Author: Rui Fan <19...@gmail.com>
AuthorDate: Fri Feb 23 23:45:23 2024 +0800
[FLINK-34502][autoscaler] Support calculating network memory for forward and rescale edge (#782)
---
.../flink/autoscaler/ScalingMetricEvaluator.java | 4 +-
.../flink/autoscaler/topology/JobTopology.java | 22 +++++----
.../flink/autoscaler/topology/VertexInfo.java | 14 +++---
.../flink/autoscaler/tuning/MemoryTuning.java | 56 ++++++++++++++++++----
.../flink/autoscaler/BacklogBasedScalingTest.java | 22 +++++----
.../flink/autoscaler/JobAutoScalerImplTest.java | 3 +-
.../MetricsCollectionAndEvaluationTest.java | 26 +++++-----
.../autoscaler/RecommendedParallelismTest.java | 10 ++--
.../flink/autoscaler/ScalingExecutorTest.java | 5 +-
.../autoscaler/ScalingMetricCollectorTest.java | 29 ++++++-----
.../autoscaler/ScalingMetricEvaluatorTest.java | 28 ++++++-----
.../flink/autoscaler/ScalingTrackingTest.java | 3 +-
.../autoscaler/metrics/ScalingMetricsTest.java | 11 +++--
.../flink/autoscaler/topology/JobTopologyTest.java | 25 ++++++----
.../{utils => tuning}/MemoryTuningTest.java | 38 ++++++++++++---
15 files changed, 194 insertions(+), 102 deletions(-)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
index 6512758b..085d80ae 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
@@ -338,7 +338,7 @@ public class ScalingMetricEvaluator {
}
out.put(CATCH_UP_DATA_RATE, EvaluatedScalingMetric.of(catchUpInputRate));
} else {
- var inputs = topology.get(vertex).getInputs();
+ var inputs = topology.get(vertex).getInputs().keySet();
double sumAvgTargetRate = 0;
double sumCatchUpDataRate = 0;
for (var inputVertex : inputs) {
@@ -531,7 +531,7 @@ public class ScalingMetricEvaluator {
JobVertexID from,
JobVertexID to) {
- var toVertexInputs = topology.get(to).getInputs();
+ var toVertexInputs = topology.get(to).getInputs().keySet();
// Case 1: Downstream vertex has single input (from) so we can use the most reliable num
// records in
if (toVertexInputs.size() == 1) {
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 c0a98d2e..c140c423 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
@@ -61,7 +61,7 @@ public class JobTopology {
public JobTopology(Set<VertexInfo> vertexInfo) {
- Map<JobVertexID, Set<JobVertexID>> vertexOutputs = new HashMap<>();
+ Map<JobVertexID, Map<JobVertexID, String>> vertexOutputs = new HashMap<>();
vertexInfos =
ImmutableMap.copyOf(
vertexInfo.stream().collect(Collectors.toMap(VertexInfo::getId, v -> v)));
@@ -72,13 +72,13 @@ public class JobTopology {
info -> {
var vertexId = info.getId();
- vertexOutputs.computeIfAbsent(vertexId, id -> new HashSet<>());
+ vertexOutputs.computeIfAbsent(vertexId, id -> new HashMap<>());
info.getInputs()
.forEach(
- inputId ->
+ (inputId, shipStrategy) ->
vertexOutputs
- .computeIfAbsent(inputId, id -> new HashSet<>())
- .add(vertexId));
+ .computeIfAbsent(inputId, id -> new HashMap<>())
+ .put(vertexId, shipStrategy));
if (info.isFinished()) {
finishedVertices.add(vertexId);
}
@@ -105,7 +105,8 @@ public class JobTopology {
List<JobVertexID> sorted = new ArrayList<>(vertexInfos.size());
Map<JobVertexID, List<JobVertexID>> remainingInputs = new HashMap<>(vertexInfos.size());
- vertexInfos.forEach((id, v) -> remainingInputs.put(id, new ArrayList<>(v.getInputs())));
+ vertexInfos.forEach(
+ (id, v) -> remainingInputs.put(id, new ArrayList<>(v.getInputs().keySet())));
while (!remainingInputs.isEmpty()) {
List<JobVertexID> verticesWithZeroIndegree = new ArrayList<>();
@@ -122,6 +123,7 @@ public class JobTopology {
vertexInfos
.get(v)
.getOutputs()
+ .keySet()
.forEach(o -> remainingInputs.get(o).remove(v));
});
@@ -143,20 +145,22 @@ public class JobTopology {
for (JsonNode node : nodes) {
var vertexId = JobVertexID.fromHexString(node.get("id").asText());
- var inputList = new HashSet<JobVertexID>();
+ var inputs = new HashMap<JobVertexID, String>();
var ioMetrics = metrics.get(vertexId);
var finished = finishedVertices.contains(vertexId);
vertexInfo.add(
new VertexInfo(
vertexId,
- inputList,
+ inputs,
node.get("parallelism").asInt(),
maxParallelismMap.get(vertexId),
finished,
finished ? IOMetrics.FINISHED_METRICS : ioMetrics));
if (node.has("inputs")) {
for (JsonNode input : node.get("inputs")) {
- inputList.add(JobVertexID.fromHexString(input.get("id").asText()));
+ inputs.put(
+ JobVertexID.fromHexString(input.get("id").asText()),
+ input.get("ship_strategy").asText());
}
}
}
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 3945ab7e..69e669a3 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
@@ -22,7 +22,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import lombok.Data;
-import java.util.Set;
+import java.util.Map;
/** Job vertex information. */
@Data
@@ -30,9 +30,11 @@ public class VertexInfo {
private final JobVertexID id;
- private final Set<JobVertexID> inputs;
+ // All input vertices and the ship_strategy
+ private final Map<JobVertexID, String> inputs;
- private Set<JobVertexID> outputs;
+ // All output vertices and the ship_strategy
+ private Map<JobVertexID, String> outputs;
private final int parallelism;
@@ -46,7 +48,7 @@ public class VertexInfo {
public VertexInfo(
JobVertexID id,
- Set<JobVertexID> inputs,
+ Map<JobVertexID, String> inputs,
int parallelism,
int maxParallelism,
boolean finished,
@@ -63,7 +65,7 @@ public class VertexInfo {
@VisibleForTesting
public VertexInfo(
JobVertexID id,
- Set<JobVertexID> inputs,
+ Map<JobVertexID, String> inputs,
int parallelism,
int maxParallelism,
IOMetrics ioMetrics) {
@@ -72,7 +74,7 @@ public class VertexInfo {
@VisibleForTesting
public VertexInfo(
- JobVertexID id, Set<JobVertexID> inputs, int parallelism, int maxParallelism) {
+ JobVertexID id, Map<JobVertexID, String> inputs, int parallelism, int maxParallelism) {
this(id, inputs, parallelism, maxParallelism, null);
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/MemoryTuning.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/MemoryTuning.java
index af5fd8a1..428f7a66 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/MemoryTuning.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/MemoryTuning.java
@@ -17,6 +17,7 @@
package org.apache.flink.autoscaler.tuning;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.ScalingSummary;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
@@ -243,9 +244,9 @@ public class MemoryTuning {
Configuration config,
MemoryBudget memBudget) {
- final long buffersPerChannel =
+ final int buffersPerChannel =
config.get(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL);
- final long floatingBuffers =
+ final int floatingBuffers =
config.get(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
final long memorySegmentBytes =
config.get(TaskManagerOptions.MEMORY_SEGMENT_SIZE).getBytes();
@@ -253,18 +254,30 @@ public class MemoryTuning {
long maxNetworkMemory = 0;
for (VertexInfo vertexInfo : jobTopology.getVertexInfos().values()) {
// Add max amount of memory for each input gate
- for (JobVertexID input : vertexInfo.getInputs()) {
- int inputParallelism = updatedParallelisms.get(input);
+ for (Map.Entry<JobVertexID, String> inputEntry : vertexInfo.getInputs().entrySet()) {
+ final JobVertexID inputVertexId = inputEntry.getKey();
+ final String shipStrategy = inputEntry.getValue();
maxNetworkMemory +=
- (inputParallelism * buffersPerChannel + floatingBuffers)
+ calculateNetworkSegmentNumber(
+ updatedParallelisms.get(vertexInfo.getId()),
+ updatedParallelisms.get(inputVertexId),
+ shipStrategy,
+ buffersPerChannel,
+ floatingBuffers)
* memorySegmentBytes;
}
// Add max amount of memory for each output gate
// Usually, there is just one output per task
- for (JobVertexID output : vertexInfo.getOutputs()) {
- int downstreamParallelism = updatedParallelisms.get(output);
+ for (Map.Entry<JobVertexID, String> outputEntry : vertexInfo.getOutputs().entrySet()) {
+ final JobVertexID outputVertexId = outputEntry.getKey();
+ final String shipStrategy = outputEntry.getValue();
maxNetworkMemory +=
- (downstreamParallelism * buffersPerChannel + floatingBuffers)
+ calculateNetworkSegmentNumber(
+ updatedParallelisms.get(vertexInfo.getId()),
+ updatedParallelisms.get(outputVertexId),
+ shipStrategy,
+ buffersPerChannel,
+ floatingBuffers)
* memorySegmentBytes;
}
}
@@ -277,6 +290,33 @@ public class MemoryTuning {
return new MemorySize(memBudget.budget(maxNetworkMemory));
}
+ /**
+ * Calculate how many network segment current vertex needs.
+ *
+ * @param currentVertexParallelism The parallelism of current vertex.
+ * @param connectedVertexParallelism The parallelism of connected vertex.
+ */
+ @VisibleForTesting
+ static int calculateNetworkSegmentNumber(
+ int currentVertexParallelism,
+ int connectedVertexParallelism,
+ String shipStrategy,
+ int buffersPerChannel,
+ int floatingBuffers) {
+ // TODO When the parallelism is changed via the rescale api, the FORWARD may be changed to
+ // RESCALE. This logic may needs to be updated after FLINK-33123.
+ if (currentVertexParallelism == connectedVertexParallelism
+ && "FORWARD".equals(shipStrategy)) {
+ return buffersPerChannel + floatingBuffers;
+ } else if ("FORWARD".equals(shipStrategy) || "RESCALE".equals(shipStrategy)) {
+ final int channelCount =
+ (int) Math.ceil(connectedVertexParallelism / (double) currentVertexParallelism);
+ return channelCount * buffersPerChannel + floatingBuffers;
+ } else {
+ return connectedVertexParallelism * buffersPerChannel + floatingBuffers;
+ }
+ }
+
private static MemorySize getUsage(
ScalingMetric scalingMetric, Map<ScalingMetric, EvaluatedScalingMetric> globalMetrics) {
MemorySize heapUsed = new MemorySize((long) globalMetrics.get(scalingMetric).getAverage());
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
index 0e6f84d5..10490e7b 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
@@ -40,7 +40,7 @@ import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
-import java.util.Set;
+import java.util.Map;
import java.util.SortedMap;
import static org.apache.flink.autoscaler.JobAutoScalerImpl.AUTOSCALER_ERROR;
@@ -78,9 +78,13 @@ public class BacklogBasedScalingTest {
metricsCollector =
new TestingMetricsCollector<>(
new JobTopology(
- new VertexInfo(source1, Set.of(), 1, 720, new IOMetrics(0, 0, 0)),
+ new VertexInfo(source1, Map.of(), 1, 720, new IOMetrics(0, 0, 0)),
new VertexInfo(
- sink, Set.of(source1), 1, 720, new IOMetrics(0, 0, 0))));
+ sink,
+ Map.of(source1, "REBALANCE"),
+ 1,
+ 720,
+ new IOMetrics(0, 0, 0))));
var defaultConf = context.getConfiguration();
defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
@@ -152,8 +156,8 @@ public class BacklogBasedScalingTest {
// Update topology to reflect updated parallelisms
metricsCollector.setJobTopology(
new JobTopology(
- new VertexInfo(source1, Set.of(), 4, 24),
- new VertexInfo(sink, Set.of(source1), 4, 720)));
+ new VertexInfo(source1, Map.of(), 4, 24),
+ new VertexInfo(sink, Map.of(source1, "REBALANCE"), 4, 720)));
metricsCollector.updateMetrics(
source1,
@@ -234,8 +238,8 @@ public class BacklogBasedScalingTest {
metricsCollector.setJobUpdateTs(now);
metricsCollector.setJobTopology(
new JobTopology(
- new VertexInfo(source1, Set.of(), 2, 24),
- new VertexInfo(sink, Set.of(source1), 2, 720)));
+ new VertexInfo(source1, Map.of(), 2, 24),
+ new VertexInfo(sink, Map.of(source1, "REBALANCE"), 2, 720)));
/* Test stability while processing backlog. */
@@ -356,8 +360,8 @@ public class BacklogBasedScalingTest {
context = context.toBuilder().jobStatus(JobStatus.RUNNING).build();
metricsCollector.setJobTopology(
new JobTopology(
- new VertexInfo(source1, Set.of(), 4, 720),
- new VertexInfo(sink, Set.of(source1), 4, 720)));
+ new VertexInfo(source1, Map.of(), 4, 720),
+ new VertexInfo(sink, Map.of(source1, "REBALANCE"), 4, 720)));
var expectedEndTime = Instant.ofEpochMilli(10);
metricsCollector.setJobUpdateTs(expectedEndTime);
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
index e3ed6625..4ec10b43 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
@@ -53,7 +53,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import java.util.TreeMap;
import static java.util.Map.entry;
@@ -87,7 +86,7 @@ public class JobAutoScalerImplTest {
@Test
void testMetricReporting() throws Exception {
JobVertexID jobVertexID = new JobVertexID();
- JobTopology jobTopology = new JobTopology(new VertexInfo(jobVertexID, Set.of(), 1, 10));
+ JobTopology jobTopology = new JobTopology(new VertexInfo(jobVertexID, Map.of(), 1, 10));
var metricsCollector =
new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(jobTopology);
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 a44b0c17..d0b0daab 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
@@ -47,7 +47,6 @@ import java.time.ZoneId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.function.Supplier;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
@@ -89,11 +88,16 @@ public class MetricsCollectionAndEvaluationTest {
topology =
new JobTopology(
- new VertexInfo(source1, Set.of(), 2, 720, new IOMetrics(0, 0, 0)),
- new VertexInfo(source2, Set.of(), 2, 720, new IOMetrics(0, 0, 0)),
+ new VertexInfo(source1, Map.of(), 2, 720, new IOMetrics(0, 0, 0)),
+ new VertexInfo(source2, Map.of(), 2, 720, new IOMetrics(0, 0, 0)),
new VertexInfo(
- map, Set.of(source1, source2), 12, 720, new IOMetrics(0, 0, 0)),
- new VertexInfo(sink, Set.of(map), 8, 24, new IOMetrics(0, 0, 0)));
+ map,
+ Map.of(source1, "REBALANCE", source2, "REBALANCE"),
+ 12,
+ 720,
+ new IOMetrics(0, 0, 0)),
+ new VertexInfo(
+ sink, Map.of(map, "REBALANCE"), 8, 24, new IOMetrics(0, 0, 0)));
metricsCollector = new TestingMetricsCollector<>(topology);
@@ -335,7 +339,7 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testTolerateAbsenceOfPendingRecordsMetric() throws Exception {
- var topology = new JobTopology(new VertexInfo(source1, Set.of(), 5, 720));
+ var topology = new JobTopology(new VertexInfo(source1, Map.of(), 5, 720));
metricsCollector = new TestingMetricsCollector(topology);
metricsCollector.setJobUpdateTs(startTime);
@@ -408,9 +412,9 @@ public class MetricsCollectionAndEvaluationTest {
var finished = new JobVertexID();
var topology =
new JobTopology(
- new VertexInfo(s1, Set.of(), 10, 720),
+ new VertexInfo(s1, Map.of(), 10, 720),
new VertexInfo(
- finished, Set.of(), 10, 720, true, IOMetrics.FINISHED_METRICS));
+ finished, Map.of(), 10, 720, true, IOMetrics.FINISHED_METRICS));
metricsCollector = new TestingMetricsCollector(topology);
metricsCollector.setJobUpdateTs(startTime);
@@ -445,7 +449,7 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testObservedTprCollection() throws Exception {
var source = new JobVertexID();
- var topology = new JobTopology(new VertexInfo(source, Set.of(), 10, 720));
+ var topology = new JobTopology(new VertexInfo(source, Map.of(), 10, 720));
metricsCollector = new TestingMetricsCollector(topology);
metricsCollector.setJobUpdateTs(startTime);
@@ -521,7 +525,7 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testMetricCollectionDuringStabilization() throws Exception {
var source = new JobVertexID();
- var topology = new JobTopology(new VertexInfo(source, Set.of(), 10, 720));
+ var topology = new JobTopology(new VertexInfo(source, Map.of(), 10, 720));
metricsCollector = new TestingMetricsCollector(topology);
metricsCollector.setJobUpdateTs(startTime);
@@ -596,7 +600,7 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testScaleDownWithZeroProcessingRate() throws Exception {
- var topology = new JobTopology(new VertexInfo(source1, Set.of(), 2, 720));
+ var topology = new JobTopology(new VertexInfo(source1, Map.of(), 2, 720));
metricsCollector = new TestingMetricsCollector<>(topology);
metricsCollector.setJobUpdateTs(startTime);
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
index 733d5273..9b225ba6 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
@@ -37,7 +37,7 @@ import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
-import java.util.Set;
+import java.util.Map;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.getRestClusterClientSupplier;
@@ -74,8 +74,8 @@ public class RecommendedParallelismTest {
metricsCollector =
new TestingMetricsCollector<>(
new JobTopology(
- new VertexInfo(source, Set.of(), 1, 720),
- new VertexInfo(sink, Set.of(source), 1, 720)));
+ new VertexInfo(source, Map.of(), 1, 720),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 720)));
var defaultConf = context.getConfiguration();
defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
@@ -198,8 +198,8 @@ public class RecommendedParallelismTest {
// updating the topology to reflect the scale
metricsCollector.setJobTopology(
new JobTopology(
- new VertexInfo(source, Set.of(), 4, 24),
- new VertexInfo(sink, Set.of(source), 4, 720)));
+ new VertexInfo(source, Map.of(), 4, 24),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 4, 720)));
now = now.plus(Duration.ofSeconds(10));
setClocksTo(now);
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 afb6773e..46b32424 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
@@ -45,7 +45,6 @@ import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
@@ -318,8 +317,8 @@ public class ScalingExecutorTest {
JobTopology jobTopology =
new JobTopology(
- new VertexInfo(source, Set.of(), 10, 1000, false, null),
- new VertexInfo(sink, Set.of(source), 10, 1000, false, null));
+ new VertexInfo(source, Map.of(), 10, 1000, false, null),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 10, 1000, false, null));
assertTrue(
scalingDecisionExecutor.scaleResource(
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 6ad91f63..6157e487 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
@@ -217,10 +217,15 @@ public class ScalingMetricCollectorTest {
var sourceId = JobVertexID.fromHexString("bc764cd8ddf7a0cff126f51c16239658");
var sinkId = JobVertexID.fromHexString("20ba6b65f97481d5570070de90e4e791");
- var source = new VertexInfo(sourceId, Set.of(), 2, 128, true, IOMetrics.FINISHED_METRICS);
+ var source = new VertexInfo(sourceId, Map.of(), 2, 128, true, IOMetrics.FINISHED_METRICS);
var sink =
new VertexInfo(
- sinkId, Set.of(sourceId), 2, 128, false, new IOMetrics(291532, 1, 2));
+ sinkId,
+ Map.of(sourceId, "HASH"),
+ 2,
+ 128,
+ false,
+ new IOMetrics(291532, 1, 2));
assertEquals(
new JobTopology(source, sink), metricsCollector.getJobTopology(jobDetailsInfo));
@@ -269,13 +274,13 @@ public class ScalingMetricCollectorTest {
var t1 =
new JobTopology(
- new VertexInfo(source, Set.of(), 1, 1),
- new VertexInfo(sink, Set.of(source), 1, 1));
+ new VertexInfo(source, Map.of(), 1, 1),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 1));
var t2 =
new JobTopology(
- new VertexInfo(source2, Set.of(), 1, 1),
- new VertexInfo(sink, Set.of(source2), 1, 1));
+ new VertexInfo(source2, Map.of(), 1, 1),
+ new VertexInfo(sink, Map.of(source2, "REBALANCE"), 1, 1));
collector.queryFilteredMetricNames(context, t1);
assertEquals(1, metricNameQueryCounter.get(source));
@@ -303,8 +308,8 @@ public class ScalingMetricCollectorTest {
// Mark source finished, should not be queried again
t2 =
new JobTopology(
- new VertexInfo(source2, Set.of(), 1, 1, true, null),
- new VertexInfo(sink, Set.of(source2), 1, 1));
+ new VertexInfo(source2, Map.of(), 1, 1, true, null),
+ new VertexInfo(sink, Map.of(source2, "REBALANCE"), 1, 1));
collector.queryFilteredMetricNames(context, t2);
assertEquals(3, metricNameQueryCounter.get(source));
assertEquals(2, metricNameQueryCounter.get(source2));
@@ -327,8 +332,8 @@ public class ScalingMetricCollectorTest {
var sink = new JobVertexID();
var topology =
new JobTopology(
- new VertexInfo(source, Set.of(), 1, 1),
- new VertexInfo(sink, Set.of(source), 1, 1));
+ new VertexInfo(source, Map.of(), 1, 1),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 1));
testRequiredMetrics(
metricList, getSourceRequiredMetrics(), testCollector, source, topology);
@@ -404,8 +409,8 @@ public class ScalingMetricCollectorTest {
var sink = new JobVertexID();
var topology =
new JobTopology(
- new VertexInfo(source, Set.of(), 1, 1),
- new VertexInfo(sink, Set.of(source), 1, 1));
+ new VertexInfo(source, Map.of(), 1, 1),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 1));
var metricCollector = new TestingMetricsCollector<>(topology);
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
index bfba0c0b..c4559f14 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
@@ -36,7 +36,6 @@ import java.time.Instant;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import java.util.TreeMap;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.CATCH_UP_DURATION;
@@ -60,8 +59,8 @@ public class ScalingMetricEvaluatorTest {
var topology =
new JobTopology(
- new VertexInfo(source, Collections.emptySet(), 1, 1, null),
- new VertexInfo(sink, Set.of(source), 1, 1, null));
+ new VertexInfo(source, Collections.emptyMap(), 1, 1, null),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 1, null));
var metricHistory = new TreeMap<Instant, CollectedMetrics>();
@@ -324,8 +323,8 @@ public class ScalingMetricEvaluatorTest {
var topology =
new JobTopology(
- new VertexInfo(source, Collections.emptySet(), 1, 1),
- new VertexInfo(sink, Set.of(source), 1, 1));
+ new VertexInfo(source, Collections.emptyMap(), 1, 1),
+ new VertexInfo(sink, Map.of(source, "REBALANCE"), 1, 1));
var metricHistory = new TreeMap<Instant, CollectedMetrics>();
@@ -660,10 +659,11 @@ public class ScalingMetricEvaluatorTest {
var topology =
new JobTopology(
- new VertexInfo(source1, Collections.emptySet(), 1, 1),
- new VertexInfo(source2, Collections.emptySet(), 1, 1),
- new VertexInfo(op1, Set.of(source1, source2), 1, 1),
- new VertexInfo(sink1, Set.of(op1), 1, 1));
+ new VertexInfo(source1, Collections.emptyMap(), 1, 1),
+ new VertexInfo(source2, Collections.emptyMap(), 1, 1),
+ new VertexInfo(
+ op1, Map.of(source1, "REBALANCE", source2, "REBALANCE"), 1, 1),
+ new VertexInfo(sink1, Map.of(op1, "REBALANCE"), 1, 1));
var metricHistory = new TreeMap<Instant, CollectedMetrics>();
@@ -726,10 +726,12 @@ public class ScalingMetricEvaluatorTest {
var topology =
new JobTopology(
- new VertexInfo(source1, Collections.emptySet(), 1, 1),
- new VertexInfo(source2, Collections.emptySet(), 1, 1),
- new VertexInfo(op1, Set.of(source1, source2), 1, 1),
- new VertexInfo(op2, Set.of(source1, source2), 1, 1));
+ new VertexInfo(source1, Collections.emptyMap(), 1, 1),
+ new VertexInfo(source2, Collections.emptyMap(), 1, 1),
+ new VertexInfo(
+ op1, Map.of(source1, "REBALANCE", source2, "REBALANCE"), 1, 1),
+ new VertexInfo(
+ op2, Map.of(source1, "REBALANCE", source2, "REBALANCE"), 1, 1));
var metricHistory = new TreeMap<Instant, CollectedMetrics>();
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingTrackingTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingTrackingTest.java
index 85cb65d8..2a9d8ae1 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingTrackingTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingTrackingTest.java
@@ -209,8 +209,7 @@ class ScalingTrackingTest {
Set<VertexInfo> vertexInfos = new HashSet<>();
for (Map.Entry<JobVertexID, Integer> entry : parallelisms.entrySet()) {
vertexInfos.add(
- new VertexInfo(
- entry.getKey(), new HashSet<>(), entry.getValue(), entry.getValue()));
+ new VertexInfo(entry.getKey(), Map.of(), entry.getValue(), entry.getValue()));
}
return vertexInfos;
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/ScalingMetricsTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/ScalingMetricsTest.java
index c689a0a9..d858a1e9 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/ScalingMetricsTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/ScalingMetricsTest.java
@@ -33,7 +33,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -52,8 +51,9 @@ public class ScalingMetricsTest {
var topology =
new JobTopology(
new VertexInfo(
- source, Collections.emptySet(), 1, 1, new IOMetrics(1, 2, 3)),
- new VertexInfo(op, Set.of(source), 1, 1, new IOMetrics(1, 2, 3)));
+ source, Collections.emptyMap(), 1, 1, new IOMetrics(1, 2, 3)),
+ new VertexInfo(
+ op, Map.of(source, "REBALANCE"), 1, 1, new IOMetrics(1, 2, 3)));
Map<ScalingMetric, Double> scalingMetrics = new HashMap<>();
ScalingMetrics.computeDataRateMetrics(
@@ -221,8 +221,9 @@ public class ScalingMetricsTest {
var topology =
new JobTopology(
new VertexInfo(
- SOURCE, Collections.emptySet(), 1, 1, new IOMetrics(0, 0, 0)),
- new VertexInfo(sink, Set.of(SOURCE), 1, 1, new IOMetrics(0, 0, 0)));
+ SOURCE, Collections.emptyMap(), 1, 1, new IOMetrics(0, 0, 0)),
+ new VertexInfo(
+ sink, Map.of(SOURCE, "REBALANCE"), 1, 1, new IOMetrics(0, 0, 0)));
Map<ScalingMetric, Double> scalingMetrics = new HashMap<>();
scalingMetrics.put(ScalingMetric.LAG, lag);
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 cdbc203b..e38295e1 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
@@ -30,7 +30,6 @@ import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -45,17 +44,17 @@ public class JobTopologyTest {
var s2 = env.fromElements(1).name("s2");
s1.union(s2)
- .shuffle()
+ .keyBy(v -> v)
.map(i -> i)
.name("map1")
.setParallelism(2)
- .shuffle()
.print()
+ .disableChaining()
.name("sink1")
- .setParallelism(3);
+ .setParallelism(2);
var s3 = env.fromElements(1).name("s3");
- var map2 = s3.shuffle().map(i -> i).name("map2").setParallelism(4).shuffle();
+ var map2 = s3.rescale().map(i -> i).name("map2").setParallelism(4).rebalance();
map2.print().name("sink2").setParallelism(5);
map2.print().name("sink3").setParallelism(6);
@@ -83,17 +82,25 @@ public class JobTopologyTest {
assertTrue(jobTopology.get(vertices.get("Sink: sink3")).getOutputs().isEmpty());
assertEquals(
- Set.of(vertices.get("map1")),
+ Map.of(vertices.get("map1"), "HASH"),
jobTopology.get(vertices.get("Source: s1")).getOutputs());
assertEquals(
- Set.of(vertices.get("map1")),
+ Map.of(vertices.get("map1"), "HASH"),
jobTopology.get(vertices.get("Source: s2")).getOutputs());
assertEquals(
- Set.of(vertices.get("map2")),
+ Map.of(vertices.get("Sink: sink1"), "FORWARD"),
+ jobTopology.get(vertices.get("map1")).getOutputs());
+
+ assertEquals(
+ Map.of(vertices.get("map2"), "RESCALE"),
jobTopology.get(vertices.get("Source: s3")).getOutputs());
assertEquals(
- Set.of(vertices.get("Sink: sink2"), vertices.get("Sink: sink3")),
+ Map.of(
+ vertices.get("Sink: sink2"),
+ "REBALANCE",
+ vertices.get("Sink: sink3"),
+ "REBALANCE"),
jobTopology.get(vertices.get("map2")).getOutputs());
assertEquals(2, jobTopology.get(vertices.get("map1")).getParallelism());
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/MemoryTuningTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
similarity index 88%
rename from flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/MemoryTuningTest.java
rename to flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
index 2838e228..b3c99716 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/MemoryTuningTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.autoscaler.utils;
+package org.apache.flink.autoscaler.tuning;
import org.apache.flink.api.common.JobID;
import org.apache.flink.autoscaler.JobAutoScalerContext;
@@ -28,8 +28,6 @@ import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.autoscaler.topology.VertexInfo;
-import org.apache.flink.autoscaler.tuning.ConfigChanges;
-import org.apache.flink.autoscaler.tuning.MemoryTuning;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -39,7 +37,6 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
@@ -87,8 +84,14 @@ public class MemoryTuningTest {
JobTopology jobTopology =
new JobTopology(
- new VertexInfo(jobVertex1, Set.of(), 50, 1000, false, null),
- new VertexInfo(jobVertex2, Set.of(jobVertex1), 50, 1000, false, null));
+ new VertexInfo(jobVertex1, Map.of(), 50, 1000, false, null),
+ new VertexInfo(
+ jobVertex2,
+ Map.of(jobVertex1, "REBALANCE"),
+ 50,
+ 1000,
+ false,
+ null));
Map<JobVertexID, ScalingSummary> scalingSummaries =
Map.of(
@@ -195,4 +198,27 @@ public class MemoryTuningTest {
.startsWith(
"Memory tuning recommends the following configuration (automatic tuning is disabled):");
}
+
+ @Test
+ void testCalculateNetworkSegmentNumber() {
+ // Test FORWARD
+ assertThat(MemoryTuning.calculateNetworkSegmentNumber(10, 10, "FORWARD", 2, 8))
+ .isEqualTo(10);
+
+ // Test FORWARD is changed to RESCALE.
+ assertThat(MemoryTuning.calculateNetworkSegmentNumber(10, 15, "FORWARD", 2, 8))
+ .isEqualTo(12);
+
+ // Test RESCALE.
+ assertThat(MemoryTuning.calculateNetworkSegmentNumber(10, 15, "RESCALE", 2, 8))
+ .isEqualTo(12);
+
+ // Test REBALANCE.
+ assertThat(MemoryTuning.calculateNetworkSegmentNumber(10, 15, "REBALANCE", 2, 8))
+ .isEqualTo(38);
+
+ // Test Unrecognizable.
+ assertThat(MemoryTuning.calculateNetworkSegmentNumber(10, 15, "Unrecognizable", 2, 8))
+ .isEqualTo(38);
+ }
}