title: Flink Job not getting submitted. java.io.IOException: Cannot allocate memory
body: <p>I am using Flink session cluster(Kubernetes Cluster (Session Mode)) to deploy batch jobs with HA. Inside the recovery/default/blob/ directory, directories starting with job_ is getting piled up.</p>
  <pre><code>drwxr-xr-x 1 flink flink 1 Nov 16 09:03 job_747a694a765d1b580a703e2785a9e3fa
  </code></pre>
  <p>Job get submitted every 1 min. But in the above above ls -ltr of /recovery/default/blob/, blobs of one of the job is not getting cleared. This job has neither completed nor failed. Also it is not listed in the list on the web ui</p>
  <p>The log file when this happens is</p>
  <pre><code>2021-11-22 09:03:11,537 INFO  org.apache.flink.kubernetes.highavailability.KubernetesHaServices [] - Finished cleaning up the high availability data for job 6a71a36a3c82d8a9438c9aa9ed6b8993.
  2021-11-22 09:03:14,904 ERROR org.apache.flink.runtime.blob.BlobServerConnection           [] - PUT operation failed
  java.io.IOException: Cannot allocate memory
      at java.io.FileOutputStream.writeBytes(Native Method) ~[?:1.8.0_312]
      at java.io.FileOutputStream.write(FileOutputStream.java:326) ~[?:1.8.0_312]
      at org.apache.flink.core.fs.local.LocalDataOutputStream.write(LocalDataOutputStream.java:55) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.shaded.guava30.com.google.common.io.ByteStreams.copy(ByteStreams.java:113) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.shaded.guava30.com.google.common.io.ByteSource.copyTo(ByteSource.java:243) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.shaded.guava30.com.google.common.io.Files.copy(Files.java:301) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.FileSystemBlobStore.put(FileSystemBlobStore.java:79) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.FileSystemBlobStore.put(FileSystemBlobStore.java:72) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.BlobUtils.moveTempFileToStore(BlobUtils.java:385) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.BlobServer.moveTempFileToStore(BlobServer.java:680) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.BlobServerConnection.put(BlobServerConnection.java:350) [flink-dist_2.11-1.14.0.jar:1.14.0]
      at org.apache.flink.runtime.blob.BlobServerConnection.run(BlobServerConnection.java:110) [flink-dist_2.11-1.14.0.jar:1.14.0]
  </code></pre>
  <p>It seems that Flink is not retrying this job. Is there a config which can retry this job?</p>