title: How to catch FlinkRuntimeException in Flink producer and skip failed messages
body: <p>Is there any way to catch FlinkRuntimeException and handle it properly preventing Flink job restating?</p>
  <p><strong>Code sample which throws the FlinkRuntimeException:</strong></p>
  <pre><code>    int numberOfRows = 1;
      int rowsPerSecond = 1;
  
      DataStream&lt;String&gt; stream = environment.addSource(
                      new DataGeneratorSource&lt;&gt;(
                              RandomGenerator.stringGenerator(1050000), // max.message.bytes=1048588
                              rowsPerSecond,
                              (long) numberOfRows),
                      TypeInformation.of(String.class))
              .setParallelism(1)
              .name(&quot;string-generator&quot;);
  
  
      KafkaSinkBuilder&lt;String&gt; builder = KafkaSink.&lt;String&gt;builder()
              .setBootstrapServers(&quot;localhost:9092&quot;)
              .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
              .setRecordSerializer(
                      KafkaRecordSerializationSchema.builder().setTopic(&quot;test.output&quot;)
                              .setValueSerializationSchema(new SimpleStringSchema())
                              .build());
  
  
      KafkaSink&lt;String&gt; sink = builder.build();
  
      stream.sinkTo(sink).setParallelism(1).name(&quot;output-producer&quot;);
  </code></pre>
  <p><strong>Exception Stack Trace:</strong></p>
  <blockquote>
  <p>2022-06-02/14:01:45.066/PDT [flink-akka.actor.default-dispatcher-4] INFO  output-producer: Writer -&gt; output-producer: Committer (1/1) (a66beca5a05c1c27691f7b94ca6ac025)
  switched from RUNNING to FAILED on 271b1b90-7d6b-4a34-8116-3de6faa8a9bf @ 127.0.0.1 (dataPort=-1).
  org.apache.flink.util.FlinkRuntimeException: Failed to send data to Kafka null with FlinkKafkaInternalProducer{transactionalId='null', inTransaction=false, closed=false}
  at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.throwException(KafkaWriter.java:440) ~[flink-connector-kafka-1.15.0.jar:1.15.0]
  at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.lambda$onCompletion$0(KafkaWriter.java:421) ~[flink-connector-kafka-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsNonBlocking(MailboxProcessor.java:353) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:317) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:804) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:753) ~[flink-streaming-java-1.15.0.jar:1.15.0]
  at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948) ~[flink-runtime-1.15.0.jar:1.15.0]
  at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927) ~[flink-runtime-1.15.0.jar:1.15.0]
  at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741) ~[flink-runtime-1.15.0.jar:1.15.0]
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563) ~[flink-runtime-1.15.0.jar:1.15.0]
  at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_292]
  Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1050088 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.</p>
  </blockquote>