Understanding Flink TaskManager Memory Allocation on YARN (Per‑Job Mode)

We submitted a Flink on YARN job (per‑job mode) with the following command:

/opt/flink-1.9.0/bin/flink run \
--detached \
--jobmanager yarn-cluster \
--yarnname "x.y.z" \
--yarnjobManagerMemory 2048 \
--yarntaskManagerMemory 4096 \
--yarnslots 2 \
--parallelism 20 \
--class x.y.z \
xyz-1.0.jar

The job started 10 TaskManagers and ran normally. Opening a TaskManager UI page revealed that although we set the TaskManager memory to 4 GB, the JVM heap shown was only 2.47 GB, with an additional "Flink Managed Memory" of 1.78 GB.

VisualVM confirmed that the JVM parameters were set to -Xms and -Xmx matching the calculated heap size.

TaskManager memory layout consists of:

Network Buffer (off‑heap since Flink 1.5) – controlled by taskmanager.network.memory.fraction, taskmanager.network.memory.min, and taskmanager.network.memory.max.

# Default network buffer fraction (0.1)
taskmanager.network.memory.fraction: 0.15
# Min and max values
taskmanager.network.memory.min: 128mb
taskmanager.network.memory.max: 1gb

Flink Managed Memory – used by internal operators, allocated on‑heap by default; its size is governed by taskmanager.memory.fraction and taskmanager.memory.off-heap.

# Fraction of heap for managed memory (default 0.7)
taskmanager.memory.fraction: 0.7
# Off‑heap managed memory disabled by default
taskmanager.memory.off-heap: false

Free memory – the remaining heap after subtracting managed memory.

To understand the numbers, we examined the Flink 1.9.0 source. The YARN per‑job cluster entry point is org.apache.flink.yarn.YarnClusterDescriptor, which creates a ClusterSpecification containing JobManager memory, TaskManager memory, number of TaskManagers, and slots per TaskManager.

The method ContaineredTaskManagerParameters.calculateCutoffMB computes how much container memory must be reserved for non‑TaskManager overhead, using the configuration keys containerized.heap-cutoff-ratio (default 0.25) and containerized.heap-cutoff-min (default 600 MB).

With our parameters (TaskManager memory = 4096 MB), the total usable memory is:

tm_total_memory = taskmanager.heap.size - max[containerized.heap-cutoff-min, taskmanager.heap.size * containerized.heap-cutoff-ratio]
# 4096 - max[600, 4096 * 0.25] = 3072 MB

The network buffer size is then calculated as:

network_buffer_memory = min[taskmanager.network.memory.max, max(taskmanager.network.memory.min, tm_total_memory * taskmanager.network.memory.fraction)]
# min[1024, max[128, 3072 * 0.15]] = 460.8 MB

Thus the actual heap used by the TaskManager is:

tm_heap_memory = tm_total_memory - network_buffer_memory = 3072 - 460.8 ≈ 2611 MB

Finally, the managed memory size is:

flink_managed_memory = tm_heap_memory * taskmanager.memory.fraction = 2611 * 0.7 ≈ 1827 MB

These calculations match the values observed in the TaskManager UI and VisualVM.

For completeness, the source method TaskManagerServices.createMemoryManager shows how the configured memory size is turned into a MemoryManager instance, handling both heap and off‑heap cases.