private FlinkKafkaInternalProducer<byte[], byte[]> createTransactionalProducer()
            throws FlinkKafkaException {
        String transactionalId = availableTransactionalIds.poll();
        if (transactionalId == null) {
            throw new FlinkKafkaException(
                    FlinkKafkaErrorCode.PRODUCERS_POOL_EMPTY,
                    "Too many ongoing snapshots. Increase kafka producers pool size or "
                            + "decrease number of concurrent checkpoints.");
        }
        FlinkKafkaInternalProducer<byte[], byte[]> producer =
                initTransactionalProducer(transactionalId, true);
        producer.initTransactions();
        return producer;
    }
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