title: Flink Cluster 1.13.1 Streaming Job fails with Lost leadership Issues
body: <p>Could Someone please assist me How can I come over this Lost Leadership Issue/Error as this causing Real-time jobs to miss the timelines and this happens with the Flink cluster every once a month?</p>
  <pre><code>   java.lang.Exception: Job leader for job id f7df1578fb94f7b6d8bc1b84ad4649aa lost leadership.
      at org.apache.flink.runtime.taskexecutor.TaskExecutor$JobLeaderListenerImpl.lambda$null$2(TaskExecutor.java:2189)
      at java.util.Optional.ifPresent(Optional.java:159)
      at org.apache.flink.runtime.taskexecutor.TaskExecutor$JobLeaderListenerImpl.lambda$jobManagerLostLeadership$3(TaskExecutor.java:2187)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:440)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:208)
      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)
  </code></pre>