code: @Override
  	protected FlinkKafkaProducer.KafkaTransactionState beginTransaction() throws FlinkKafkaException {
  		switch (semantic) {
  			case EXACTLY_ONCE:
  				FlinkKafkaInternalProducer<byte[], byte[]> producer = createTransactionalProducer();
  				producer.beginTransaction();
  				return new FlinkKafkaProducer.KafkaTransactionState(producer.getTransactionalId(), producer);
  			case AT_LEAST_ONCE:
  			case NONE:
  				// Do not create new producer on each beginTransaction() if it is not necessary
  				final FlinkKafkaProducer.KafkaTransactionState currentTransaction = currentTransaction();
  				if (currentTransaction != null && currentTransaction.producer != null) {
  					return new FlinkKafkaProducer.KafkaTransactionState(currentTransaction.producer);
  				}
  				return new FlinkKafkaProducer.KafkaTransactionState(initNonTransactionalProducer(true));
  			default:
  				throw new UnsupportedOperationException("Not implemented semantic");
  		}
  	}
docs: ------------------- Logic for handling checkpoint flushing -------------------------- //