<p>We use Apache Flink <code>job cluster</code> on Kubernetes that consists of one <code>Job Manager</code> and two <code>Task Managers</code> with two slots each. The cluster is deployed and configured using <code>Lightbend Cloudflow</code> framework.</p>

<p>We also use <code>RocksDB</code> state backend together with S3-compatible storage for the persistence. There are no any issues considering both <code>savepoints</code> creation from CLI. Our job consists of a few keyed states (<code>MapState</code>) and tends to be rather huge (we expect at least 150 Gb per each state). The <code>Restart Strategy</code> for the job is set to the <code>Failure Rate</code>. We use <code>Apache Kafka</code> as a source and sink throughout our jobs.</p>

<p>We currently doing some tests (mostly PoC's) and there are a few questions lingering:</p>

<p>We did some synthetic tests and passed incorrect events to the job. That lead to the <code>Exceptions</code> were thrown during the execution. Due to <code>Failure Rate</code> strategy the following steps happen: The Corrupted message from Kafka is read via source -> The Operator tries to process the event and eventually throws an <code>Exception</code> -> The Job restarts and reads <em>THE SAME</em> record from Kafka as at the step before -> The Operator fails -> The <code>Failure Rate</code> finally exceeds the given value and the job eventually stops. What should I do next? If we try to restart the job seems that it will be restored with the latests Kafka consumer state and will read the corrupted message once again, leading us back to the previously mentioned behavior? Which are the right steps to bare with such issues? And does Flink utilize any kind of so-called <code>Dead Letter Queues</code>?</p>

<p>The other question is about the checkpointing and restore mechanics. We are currently can't figure out which exceptions raised during a job execution are considered as critical and lead to the failure of the job following by automatic recovery from the latest checkpoint? As it described in the previous case, the ordinary <code>Exception</code> raised inside the job leads to continious restarts that finally followed by the job termination. We are looking for a cases to reproduce when something is happened with our cluster (<code>Job Manager</code> fails, <code>Task Manager</code> fails or something) that leads to the automatic recovery from the latest checkpoint. Any suggestions are welcomed considering such scenario in Kubernetes cluster.</p>

<p>We had sank into the Flink official documentation but didn't find any related information or possibly perceived it in the wrong way. Great thanks!</p>

<p>The approach that Flink's Kafka deserializer takes is that if the <code>deserialize</code> method returns null, then the Flink Kafka consumer will silently skip the corrupted message. And if it throws an <code>IOException</code>, the pipeline is restarted, which can lead to a fail/restart loop as you have noted.</p>
<p>This is described in the last paragraph of <a href="https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#the-deserializationschema" rel="nofollow noreferrer">this section of the docs</a>.</p>
<p>Past work and discussion on this topic can be found in <a href="https://issues.apache.org/jira/browse/FLINK-5583" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/FLINK-5583</a> and <a href="https://issues.apache.org/jira/browse/FLINK-3679" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/FLINK-3679</a>, and in <a href="https://github.com/apache/flink/pull/3314" rel="nofollow noreferrer">https://github.com/apache/flink/pull/3314</a>.</p>
<p>A dead letter queue would be a nice improvement, but I'm not aware of any effort in that direction. (Right now, side outputs from process functions are the only way to implement a dead letter queue.)</p>
