private void migrateNextTransactionalIdHindState(FunctionInitializationContext context)
            throws Exception {
        ListState<NextTransactionalIdHint> oldNextTransactionalIdHintState =
                context.getOperatorStateStore()
                        .getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
        nextTransactionalIdHintState =
                context.getOperatorStateStore()
                        .getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR_V2);

        ArrayList<NextTransactionalIdHint> oldTransactionalIdHints =
                Lists.newArrayList(oldNextTransactionalIdHintState.get());
        if (!oldTransactionalIdHints.isEmpty()) {
            nextTransactionalIdHintState.addAll(oldTransactionalIdHints);
            // clear old state
            oldNextTransactionalIdHintState.clear();
        }
    }
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