springkafka和springcloudstream允许我们创建事务生产者和处理器。我们可以在其中一个示例项目中看到该功能：https://github.com/spring-cloud/spring-cloud-stream-samples/tree/master/transaction-kafka-samples:

@Transactional
    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public PersonEvent process(PersonEvent data) {
        logger.info("Received event={}", data);
        Person person = new Person();
        person.setName(data.getName());

        if(shouldFail.get()) {
            shouldFail.set(false);
            throw new RuntimeException("Simulated network error");
        } else {
            //We fail every other request as a test
            shouldFail.set(true);
        }
        logger.info("Saving person={}", person);

        Person savedPerson = repository.save(person);

        PersonEvent event = new PersonEvent();
        event.setName(savedPerson.getName());
        event.setType("PersonSaved");
        logger.info("Sent event={}", event);
        return event;
    }

在这个节选中，有一个读Kafka主题，一个写在数据库中，另一个写到另一个Kafka主题，所有这些都是事务性的。
我想知道的是，而且我想回答的是，这在技术上是如何实现和实施的。
既然datasource和kafka不参与xa事务（两阶段提交），那么实现如何保证本地事务能够以事务方式从kafka读取、提交到数据库并写入kafka？