title: Restart of flink, resulting loss of data in HA mode
body: <p>I have microservice developed on top of flink layer as base image with ZK as HA. Details as:
  Environment:
  <code> deployed on Kubernetes with Apache flink 1.12.1 version</code></p>
  <p>Also, i have following configuration with k8s HA:
  Job Manager:</p>
  <pre><code>apiVersion: v1
  data:
    flink-conf.yaml: |
      state.backend: rocksdb
      blob.server.port: 6124
      jobmanager.rpc.address: job-manager
      jobmanager.rpc.port: 6123
      parallelism.default: 2
      queryable-state.proxy.ports: 6125
      taskmanager.rpc.port: 6122
      jobmanager.memory.process.size: 2900m
      taskmanager.memory.process.size: 2900m
      jobmanager.web.address: 0.0.0.0
      rest.address: 0.0.0.0
      rest.bind-address: 0.0.0.0
      rest.port: 8081
      taskmanager.data.port: 6121
      classloader.resolve-order: parent-first
      state.checkpoints.num-retained: 3
      execution.checkpointing.unaligned: false
      execution.checkpointing.max-concurrent-checkpoints: 2
      execution.checkpointing.interval: 60000
      high-availability: zookeeper
      high-availability.cluster-id: /flink-hs
      high-availability.jobmanager.port: 50010
      high-availability.storageDir:  s3://s3-bucket/zk/
      high-availability.zookeeper.client.acl: open
      high-availability.zookeeper.client.connection-timeout: 3000
      high-availability.zookeeper.client.session-timeout: 3000
      high-availability.zookeeper.client.retry-wait: 5000
      state.checkpoints.dir: s3://s3-bucket/checkpoints/
      state.backend.fs.checkpointdir: s3://s3-bucket/checkpoints/
      s3.path-style: true
      s3.path.style.access: true
    masters: job-manager:8081
  </code></pre>
  <p>Task Manager:</p>
  <pre><code>apiVersion: v1
  data:
    flink-conf.yaml: |
      state.backend: rocksdb
      blob.server.port: 6124
      jobmanager.rpc.port: 6123
      queryable-state.proxy.ports: 6125
      taskmanager.numberOfTaskSlots: 1
      taskmanager.rpc.port: 6122
      jobmanager.memory.process.size: 2900m
      taskmanager.memory.process.size: 2900m
      jobmanager.web.address: 0.0.0.0
      rest.address: 0.0.0.0
      rest.bind-address: 0.0.0.0
      rest.port: 8081
      taskmanager.data.port: 6121
      classloader.resolve-order: parent-first
      state.checkpoints.num-retained: 3
      execution.checkpointing.unaligned: false
      execution.checkpointing.max-concurrent-checkpoints: 2
      execution.checkpointing.interval: 60000
      high-availability: zookeeper
      high-availability.cluster-id: /flink-hs
      high-availability.jobmanager.port: 50010
      high-availability.storageDir:  s3://s3-bucket/zk/
      high-availability.zookeeper.client.acl: open
      high-availability.zookeeper.client.connection-timeout: 3000
      high-availability.zookeeper.client.session-timeout: 3000
      high-availability.zookeeper.client.retry-wait: 5000
      state.checkpoints.dir: s3://s3-bucket/checkpoints/
      state.backend.fs.checkpointdir: s3://s3-bucket/checkpoints/
      s3.path-style: true
      s3.path.style.access: true
    masters: job-manager:8081
  </code></pre>
  <p>Whenever i get following exception, job manager(JM) crashes. And once it is crashed, checkpoints restarts from 1, instead of from last # of checkpoint.</p>
  <pre><code>org.apache.flink.util.FlinkException: JobManager responsible for 00000000000000000000000000000000 lost the leadership.
  ...
  ...
  Caused by: org.apache.flink.util.FlinkException: ResourceManager leader changed to new address null
  
  </code></pre>
  <p>Also, i have TTL configured as follows:</p>
  <pre><code>final static String TTL = 45;
  public static final StateTtlConfig ttlConfig = StateTtlConfig
              .newBuilder(Time.days(Integer.parseInt(TTL)))
              .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
              .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
              .build();
  </code></pre>
  <p>To add, i have developed a streaming application which will be reading from kafka topic.</p>
  <p>Can somebody suggest, why my state is getting lost when i encountered above exception.
  Thank you in advance.</p>