protected void checkErroneous() throws FlinkKafkaException {
        Exception e = asyncException;
        if (e != null) {
            // prevent double throwing
            asyncException = null;
            throw new FlinkKafkaException(
                    FlinkKafkaErrorCode.EXTERNAL_ERROR,
                    "Failed to send data to Kafka: " + e.getMessage(),
                    e);
        }
    }
For each checkpoint we create new {@link FlinkKafkaInternalProducer} so that new transactions
will not clash with transactions created during previous checkpoints ({@code
producer.initTransactions()} assures that we obtain new producerId and epoch counters).
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaProducer.java