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/06/11 01:45:04 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-35489] Ensure METASPACE size is computed before the heap
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 c3e94aec [FLINK-35489] Ensure METASPACE size is computed before the heap
c3e94aec is described below
commit c3e94aec0c6f081a99b7467bd0bcee551b841600
Author: nicolas.fraison@datadoghq.com <ni...@datadoghq.com>
AuthorDate: Mon Jun 3 16:23:27 2024 +0200
[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
---
.../flink/autoscaler/tuning/MemoryTuning.java | 6 +++--
.../flink/autoscaler/tuning/MemoryTuningTest.java | 28 ++++++++++++++++++++++
2 files changed, 32 insertions(+), 2 deletions(-)
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 ad13cd9b..57ea353d 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 @@ public class MemoryTuning {
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 18f6e9c3..35ccb633 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 @@ public class MemoryTuningTest {
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(