We are running a Flink job that uses FlinkKafkaProducer09 as a sink with consumer checkpointing enabled. When our job runs into communication issues with our kafka cluster and throws an exception after the configured retries, our job restarts but we want to ensure at least once processing so we have setLogFailureOnly set to false, resulting in duplicate records from the last checkpoint to the exception after the job recovers and reconnects successfully.
We may not have the option to upgrade to the FlinkKafkaConsumer011 consumer, as our kafka endpoint is external. Are there any known ways to avoid or mitigate duplicates on the older versions of FlinkKafkaProducer while still ensuring at least once message processing?
