Flink NoResourceAvailableException: Could Not Acquire Minimum Resources

This is the error every team hits the first time they push a job past a single TaskManager. The job compiles, flink run returns, the Web UI shows the job in CREATED or SCHEDULED, and five minutes later it flips to FAILED with a slot-allocation exception. It is almost never a bug in your job. It is slot arithmetic.

1. The Error

The JobManager log ends the job with a variation of:

org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException:
  Could not acquire the minimum required resources.
    at org.apache.flink.runtime.scheduler.DefaultScheduler.lambda$assignResourceOrHandleError$6(...)
    ...
Caused by: java.util.concurrent.CompletionException:
  org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException:
  Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeout
    at org.apache.flink.runtime.scheduler.SharedSlotProfileRetrieverFactory...
Caused by: java.util.concurrent.TimeoutException:
  Timeout has occurred: 300000 ms

Two message shapes matter for search:

• Could not acquire the minimum required resources — the Adaptive Scheduler (reactive / application mode) gave up waiting for enough slots.
• Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeout — the Default Scheduler's slot.request.timeout (default 5 min / 300000 ms) expired before the SlotPool got enough slots.

Web UI symptom: the job sits in SCHEDULED, the Task Managers tab shows fewer Available Task Slots than the job needs, then the job fails. On Kubernetes native / the Operator you also see the ResourceManager repeatedly requesting new pods that never reach Running.

Confirmed against Flink 1.18–1.20, all deployment modes (standalone, YARN, native Kubernetes, Flink Kubernetes Operator). Defaults cited (taskmanager.numberOfTaskSlots = 1, slot.request.timeout = 5 min, jobmanager.adaptive-scheduler.resource-wait-timeout = 5 min) are from the 1.20 JobManagerOptions / ResourceManagerOptions sources.

2. How to Reproduce It

The fastest reproduction: a standalone cluster with 1 slot and a job with parallelism 4.

docker-compose.yml:

services:
  jobmanager:
    image: flink:1.20.0
    ports: ["8081:8081"]
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 1
  taskmanager:
    image: flink:1.20.0
    depends_on: [jobmanager]
    command: taskmanager
    scale: 1                      # one TM * one slot = 1 total slot
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 1

A trivial job whose default parallelism exceeds the one slot:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);           // needs 4 slots, cluster has 1
env.fromSequence(1, Long.MAX_VALUE)
   .map(x -> x * 2)
   .print();
env.execute("slot-starvation-demo");

docker compose up -d
flink run -c com.example.Job ./job.jar
flink list                       # job shows RUNNING briefly, then FAILED
docker compose logs jobmanager | grep -i "NoResourceAvailable"

Reproduction triggers to recognize in the wild:

• Parallelism > total slots. Total slots = numberOfTaskSlots × number of TaskManagers.
• Native K8s / Operator: kubernetes.taskmanager.cpu or memory request exceeds what the cluster can schedule, so requested TM pods stay Pending (kubectl get pods → Pending; kubectl describe pod → Insufficient cpu/memory). The Flink RM asked for pods; Kubernetes could not place them.
• YARN: requested container vcores/memory exceed queue capacity; NodeManager never allocates the container.
• A slot leak in a long-lived session cluster where slots from a previous job were never released (taskSlotsAvailable stuck at 0).

3. Why It Happens — Surface Level

Flink schedules tasks into slots, and a slot is a fixed, statically-partitioned share of one TaskManager (TM heap / numberOfTaskSlots). A job's required slot count is not its task count — it is the maximum parallelism of any single slot-sharing group (usually just the max operator parallelism). If that number exceeds the slots the cluster can offer, the SlotPool's bulk slot request cannot be fulfilled, the timeout fires, and the scheduler fails the job with NoResourceAvailableException.

Mechanically there are only two root causes: you did not give the cluster enough slots, or the cluster asked its resource provider (K8s/YARN) for TaskManagers and never got them. The first is arithmetic; the second is a scheduling/quota problem one layer down.

4. Why It Happens — Under the Hood

The allocation path is: Scheduler → SlotPool → DeclarativeSlotPool → ResourceManager → SlotManager. When a job is deployed, the scheduler computes an ExecutionSlotSharingGroup per shared group and asks the SlotPool for a bulk of slots — all-or-nothing within slot.request.timeout.

The key insight that trips people up is slot sharing. By default all tasks belong to the same default slot sharing group, so operators at different stages co-locate in the same slot. A job of source(4) → map(4) → sink(4) is 12 tasks but needs only 4 slots, because one slot runs one full parallel pipeline (source+map+sink for one key-space shard). This is why the required slot count equals the highest parallelism in the group, not the sum of task parallelisms. It is also why bumping one operator's parallelism can silently raise the whole job's slot demand.

What happens next depends on the resource manager:

• Standalone: the SlotManager has a fixed inventory — the slots that registered TaskManagers offer. It cannot create more. If demand > inventory, the request waits, then times out. resourcemanager.standalone.start-up-time (falls back to slot.request.timeout when unset) is the grace window for late-registering TMs.
• Active RMs (native K8s, YARN): the SlotManager computes the deficit and tells the RM to start new TaskManagers. The RM issues a pod/container request to Kubernetes/YARN. If the underlying platform can't place it (CPU/memory quota, taints, no nodes), the deficit never closes and the same timeout fires — the Flink error is a symptom, the real failure is Pending pods.

The Adaptive Scheduler (used in Reactive Mode and default in Application Mode from 1.17+) changes the wording and the timing. Instead of a hard bulk timeout it waits jobmanager.adaptive-scheduler.resource-wait-timeout (default 5 min) for the minimum required resources; if it gets some but not all of the desired slots, it waits resource-stabilization-timeout (default 10 s) then runs at reduced parallelism. Only if it can't reach the minimum does it emit "Could not acquire the minimum required resources" and fail. That is why the same underlying shortage produces two different messages depending on the scheduler.

5. The Fix

First, do the arithmetic. Required slots = max parallelism of the largest slot-sharing group. Total slots = taskmanager.numberOfTaskSlots × #TaskManagers. Make the second ≥ the first.

Standalone / session — give it more slots. Either raise slots per TM:

- taskmanager.numberOfTaskSlots: 1
+ taskmanager.numberOfTaskSlots: 4

or scale TaskManagers (docker compose up --scale taskmanager=4, or add replicas). Rule of thumb: set numberOfTaskSlots to the number of CPU cores per TaskManager so each slot gets ~1 core.

Or lower the job's demand when the cluster is fixed:

- env.setParallelism(4);
+ env.setParallelism(2);   // fits a 2-slot cluster

Native Kubernetes / Operator — the shortage is downstream. Don't just raise timeouts; make the pods schedulable. Check kubectl describe pod <tm-pod> first, then right-size the request:

# FlinkDeployment
  taskManager:
    resource:
-     cpu: 4
-     memory: "8192m"
+     cpu: 1
+     memory: "2048m"
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
+   # let RM start enough TMs to cover parallelism
+   kubernetes.taskmanager.cpu: "1.0"

Buy more time only when startup is genuinely slow (cold node pools, image pulls). Raising the wait window turns a fast failure into a slow one — use sparingly:

# Default Scheduler
+ slot.request.timeout: 600000            # 10 min, was 300000
# Adaptive Scheduler (reactive / application mode)
+ jobmanager.adaptive-scheduler.resource-wait-timeout: 10min

Which fix when: fixed cluster, POC → lower parallelism or raise numberOfTaskSlots. Autoscaling K8s → fix pod sizing/quotas, keep timeouts default. Slow cold starts → bump the wait timeout. Session cluster with a slot leak → cancel orphaned jobs / restart the TM; confirm taskSlotsAvailable recovers.

6. Best Practices & The Better Design

The class of problem disappears when slot demand is explicit and the cluster is sized to it, rather than discovered at deploy time.

Set slots to cores and derive parallelism from total slots:

# config.yaml (Flink 1.20 — new flattened key style)
taskmanager:
  numberOfTaskSlots: 4        # = cores per TM
parallelism:
  default: 8                  # = slots you actually provision (2 TMs * 4)

Prefer Application Mode over long-lived session clusters. Application mode runs one job per cluster with the Adaptive Scheduler, so the RM provisions exactly the TaskManagers the job needs and there is no cross-job slot contention or slot leakage. A FlinkDeployment done right:

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: payments-enricher
spec:
  flinkVersion: v1_20
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    jobmanager.adaptive-scheduler.resource-wait-timeout: "10min"
  taskManager:
    resource:
      cpu: 4
      memory: "4096m"
  job:
    jarURI: local:///opt/flink/usrlib/payments-enricher.jar
    parallelism: 8            # 2 TMs * 4 slots — provisioned exactly
    upgradeMode: savepoint

If you deliberately want operators to not share a slot (e.g. an expensive ML operator you want isolated), use .slotSharingGroup("heavy") — but know it raises total slot demand, because each named group needs its own slots. Reach for it consciously, not by accident.

7. How to Prevent It Long-Term

Alert on the metrics that predict starvation before a deploy fails:

• taskSlotsAvailable and taskSlotsTotal (RM scope) — alert when available trends to 0 on a session cluster.
• numRegisteredTaskManagers vs expected — a gap means TMs aren't registering (pod Pending, image pull, quota).
• On K8s, alert on TaskManager pods in Pending > 60s — that is the actual cause the Flink error hides.

Team conventions that stop this reaching production:

• Provision to the arithmetic: parallelism.default ≤ numberOfTaskSlots × #TMs, checked in CI against the deployment manifest.
• Slots = cores as a standing rule, so CPU sizing and slot sizing never diverge.
• Chaos test capacity: kill a TaskManager pod and confirm the job recovers within the wait timeout (ties into restart strategies — a job that can't get slots after failover throws this same exception during recovery).
• Namespace ResourceQuota on K8s sized to peak TM count, so a scale-up request can never be silently denied.

8. Key Takeaways

• Required slots = the max parallelism of the largest slot-sharing group, not the task count. Slot sharing means source→map→sink co-locate in one slot.
• Total slots = taskmanager.numberOfTaskSlots × number of TaskManagers. Make it ≥ required, or lower parallelism.
• On K8s/YARN the real failure is usually Pending TM pods/containers — right-size CPU/memory and quotas; don't just raise slot.request.timeout.
• Two messages, one cause: "Slot request bulk is not fulfillable" (Default Scheduler, 5 min) vs "Could not acquire the minimum required resources" (Adaptive Scheduler, resource-wait-timeout 5 min).
• Set slots = cores, provision parallelism to total slots, and prefer Application Mode to make demand explicit and kill the whole class of failure.