private void recycleTransactionalProducer(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
        availableTransactionalIds.add(producer.getTransactionalId());
        producer.flush();
        producer.close(Duration.ofSeconds(0));
    }
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