<p>I want to run an apache flink (1.11.1) streaming application on kubernetes. With a filesystem state backend saving to s3. Checkpointing to s3 is working</p>
<pre><code>args:
  - &quot;standalone-job&quot;
    - &quot;-s&quot;
    - &quot;s3://BUCKET_NAME/34619f2862ce3e5fc91d80eae13a434a/chk-4/_metadata&quot;
    - &quot;--job-classname&quot;
    - &quot;com.abc.def.MY_JOB&quot;
    - &quot;--kafka-broker&quot;
    - &quot;KAFKA_HOST:9092&quot;
</code></pre>
<h3>So the problem that I'm facing is:</h3>
<ul>
<li>I have to select the previous state dir manually. Is there a possibility to make it better?</li>
<li>The job increments the chk dir but it does not use the checkpoint. Means I throw a new event when I have seen an event for the first time and store it to a <code>ListState&lt;String&gt;</code> whenever I deploy via Gitlab a newer version of my application it again throws this event.</li>
<li>Why do I have to enable the checkpointing explicitly in my code when I have defined the state.backend to filesystem? <code>env.enableCheckpointing(Duration.ofSeconds(60).toMillis());</code> and <code>env.getCheckpointConfig().enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION);</code></li>
</ul>

<ul>
<li>You might be happier with <a href="https://www.ververica.com/getting-started" rel="nofollow noreferrer">Ververica Platform: Community Edition</a>, which raises the level of abstraction to the point where you don't have to deal with the details at this level. It has an API that was designed with CI/CD in mind.</li>
<li>I'm not sure I understand your second point, but it's normal that your job will rewind and reprocess some data during recovery. Flink does not guarantee exactly once processing, but rather exactly once semantics: each event will affect the state being managed by Flink exactly once. This is done by rolling back to the offsets in the most recent checkpoint, and rolling back all of the other state to what it had been after consuming all of the data up to those offsets.</li>
<li>Having a state backend is necessary as a place to store your job's working state while the job is running. If you don't enable checkpointing, then the working state won't be checkpointed, and can not be recovered. However, as of Flink 1.11, you can enable checkpointing via the config file, using</li>
</ul>
<pre><code>execution.checkpointing.interval: 60000
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
</code></pre>
