title: Flink 1.13.2: NoResourceAvailableException
body: <p>This is with Flink 1.13.2 running in Amazon's Kinesis Data Analytics Flink environment.</p>
  <p>This application is running on Kafka topics. When the topics had smaller traffic volumes, this application ran fine, with larger volumes, I'm getting this error. How do I troubleshoot and tune and fix?</p>
  <p>I see similar looking SO questions like this but that is clearly a separate issue in older versions of Flink:
  <a href="https://stackoverflow.com/questions/64252040/apache-flink-wordcount-noresourceavailableexception">Apache Flink - WordCount - NoResourceAvailableException</a></p>
  <p>The exception trace is:</p>
  <pre><code>2021-12-30 18:16:45
  java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acquire the minimum required resources.
      at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:331)
      at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:346)
      at java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:632)
      at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
      at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
      at org.apache.flink.runtime.jobmaster.slotpool.DeclarativeSlotPoolBridge$PendingRequest.failRequest(DeclarativeSlotPoolBridge.java:535)
      at org.apache.flink.runtime.jobmaster.slotpool.DeclarativeSlotPoolBridge.cancelPendingRequests(DeclarativeSlotPoolBridge.java:128)
      at org.apache.flink.runtime.jobmaster.slotpool.DeclarativeSlotPoolBridge.failPendingRequests(DeclarativeSlotPoolBridge.java:362)
      at org.apache.flink.runtime.jobmaster.slotpool.DeclarativeSlotPoolBridge.notifyNotEnoughResourcesAvailable(DeclarativeSlotPoolBridge.java:351)
      at org.apache.flink.runtime.jobmaster.JobMaster.notifyNotEnoughResourcesAvailable(JobMaster.java:824)
      at jdk.internal.reflect.GeneratedMethodAccessor133.invoke(Unknown Source)
      at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.base/java.lang.reflect.Method.invoke(Method.java:566)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:301)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)
      at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)
      at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
      at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
      at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
      at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
      at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
      at akka.actor.Actor.aroundReceive(Actor.scala:517)
      at akka.actor.Actor.aroundReceive$(Actor.scala:515)
      at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
      at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
      at akka.actor.ActorCell.invoke(ActorCell.scala:561)
      at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
      at akka.dispatch.Mailbox.run(Mailbox.scala:225)
      at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
      at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  Caused by: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acquire the minimum required resources.
      ... 30 more
  </code></pre>