Flink与Kafka集成时事务频繁失败及解决方案

def getKafkaProducer(kafkaAddr: String, targetTopicName: String, kafkaProducersPoolSize: Int): FlinkKafkaProducer[String] = {
val properties = new Properties()
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddr)
properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "180000") // 调整事务超时时间
properties.setProperty(ProducerConfig.RETRIES_CONFIG, "5") // 设置重试次数
properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "5242880") // 设置最大请求大小
properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1") // 确保每个连接上的请求数不超过1
properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") // 开启幂等性支持
val serial = new KeyedSerializationSchemaWrapper(new SimpleStringSchema())
val producer = new FlinkKafkaProducer[String](targetTopicName, serial, properties, Optional.of(new KafkaProducerPartitioner[String]()), FlinkKafkaProducer.Semantic.EXACTLY_ONCE, kafkaProducersPoolSize)
producer.setWriteTimestampToKafka(true)
producer
}

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE) // 启用每分钟一次的精确一次检查点
val cOnfig= env.getCheckpointConfig
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) // 取消作业时保留外部检查点状态
config.setMinPauseBetweenCheckpoints(3000) // 设置两次检查点之间的最短暂停时间
config.setMaxConcurrentCheckpoints(1) // 设置并行检查点的最大数量
config.setCheckpointTimeout(30000) // 设置检查点的超时时间
config.setFailOnCheckpointingErrors(false) // 设置检查点失败时不终止任务

java.lang.RuntimeException: Error while confirming checkpoint
org.apache.flink.util.FlinkRuntimeException: Committing one of transactions failed, logging first encountered failure
org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.