[FLINK-35489] Ensure METASPACE size is computed before the heap
If the heap usage with overhead leads to remaining memory to be assigned to the HEAP no
memory will be assigned to the METASPACE leading to JVM failure at startup
By setting this before the HEAP we ensure the JVM can start with appropriate amount of METASPACE memory
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 ad13cd9..57ea353 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
@@ -125,10 +125,12 @@
scalingSummaries, evaluatedMetrics.getVertexMetrics()),
config,
memBudget);
- MemorySize newHeapSize =
- determineNewSize(getUsage(HEAP_MEMORY_USED, globalMetrics), config, memBudget);
+ // Assign memory to the METASPACE before the HEAP to ensure all needed memory is provided
+ // to the METASPACE
MemorySize newMetaspaceSize =
determineNewSize(getUsage(METASPACE_MEMORY_USED, globalMetrics), config, memBudget);
+ MemorySize newHeapSize =
+ determineNewSize(getUsage(HEAP_MEMORY_USED, globalMetrics), config, memBudget);
MemorySize newManagedSize =
adjustManagedMemory(
getUsage(MANAGED_MEMORY_USED, globalMetrics),
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
index 18f6e9c..35ccb63 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/tuning/MemoryTuningTest.java
@@ -208,6 +208,34 @@
TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
"14172382822 bytes"));
+ // Test METASPACE when memory usage is so high that it could take all the memory
+ metrics = new EvaluatedMetrics(vertexMetrics, new HashMap<>(globalMetrics));
+ metrics.getGlobalMetrics()
+ .put(ScalingMetric.HEAP_MEMORY_USED, EvaluatedScalingMetric.avg(30812254720d));
+ // Set usage to max metaspace usage to ensure the calculation take in account max size
+ metrics.getGlobalMetrics()
+ .put(ScalingMetric.METASPACE_MEMORY_USED, EvaluatedScalingMetric.avg(268435456d));
+ configChanges =
+ MemoryTuning.tuneTaskManagerMemory(
+ context, metrics, jobTopology, scalingSummaries, eventHandler);
+ assertThat(configChanges.getOverrides())
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(
+ TaskManagerOptions.MANAGED_MEMORY_FRACTION.key(),
+ "0.0",
+ TaskManagerOptions.NETWORK_MEMORY_MIN.key(),
+ "13760 kb",
+ TaskManagerOptions.NETWORK_MEMORY_MAX.key(),
+ "13760 kb",
+ TaskManagerOptions.JVM_METASPACE.key(),
+ "322122547 bytes",
+ TaskManagerOptions.JVM_OVERHEAD_FRACTION.key(),
+ "0.034",
+ TaskManagerOptions.FRAMEWORK_HEAP_MEMORY.key(),
+ "0 bytes",
+ TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
+ "30 gb"));
+
// Test tuning disabled
config.set(AutoScalerOptions.MEMORY_TUNING_ENABLED, false);
assertThat(
