Spring云流数据库事务不回滚

Question

我试图编写一个spring-cloud-stream函数（spring-starter-parent 2.5.3,java 11,spring-cloud-version 2020.0.3)，该函数同时具有Kafka和Postgres事务。每当使用的消息以字符串“fail”开始时，该函数将引发一个模拟错误，我希望这将导致数据库事务回滚，然后导致kafka事务回滚。（我知道Kafka交易不是XA，这很好。）到目前为止，我还没有让数据库事务工作，但kafka事务工作。

目前我使用的是@transactional注释，它似乎不会启动数据库事务。（Kafka绑定器文档建议使用ChainedTransactionManager同步数据库+Kafka事务，但Spring Kafka文档声明不赞成使用@transactional注释，而S.C.S.针对此问题的示例使用@transaction注释和由start-jpa库创建的默认事务管理器（我认为））。我可以在调试器中看到，无论我是否@enableTransactionManagement并在我的使用者上使用@transaction，使用者都是在使用堆栈中更高的事务模板的kafka事务中执行的，但我在任何地方都看不到数据库事务。

我有几个问题想弄明白：

• 无论我是否有@transactional注释，Kafka侦听器容器都在Kafka事务的上下文中运行我的使用者，这是否正确？如果是这样，有没有办法只在Kafka事务中运行特定的函数？
• 由于容器没有方法拦截对生产者的调用（据我所知），对于生产者来说，上述情况会改变吗？
• 如何同步Kafka事务和数据库事务，以便在Kafka提交之前进行DB提交？

我有以下Crud存储库、处理程序集合和application.yml：

@Repository
public interface AuditLogRepository extends CrudRepository<AuditLog, Long> {

  /**
   * Create a new audit log entry if and only if another with the same message does not already
   * exist. This is idempotent.
   */
  @Transactional
  @Modifying
  @Query(
      nativeQuery = true,
      value = "insert into audit_log (message) values (?1) on conflict (message) do nothing")
  void createIfNotExists(String message);
}

@Profile("ft")
@Configuration
@EnableTransactionManagement
public class FaultTolerantHandlers {

  private static final Logger LOGGER = LoggerFactory.getLogger(FaultTolerantHandlers.class);

  @Bean
  public NewTopic inputTopic() {
    return TopicBuilder.name("input").partitions(1).replicas(1).build();
  }

  @Bean
  public NewTopic inputDltTopic() {
    return TopicBuilder.name("input.DLT").partitions(1).build();
  }

  @Bean
  public NewTopic leftTopic() {
    return TopicBuilder.name("left").partitions(1).build();
  }

  @Bean
  public NewTopic rightTopic() {
    return TopicBuilder.name("right").partitions(1).build();
  }

  @Bean
  public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
    return args -> {
      LOGGER.info("Producing messages to input...");
      template.send("input", "pass-1".getBytes());
      template.send("input", "fail-1".getBytes());
      template.send("input", "pass-2".getBytes());
      template.send("input", "fail-2".getBytes());
      LOGGER.info("Produced input.");
    };
  }

  @Bean
  ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(
      BinderFactory binders) {
    return (container, dest, group) -> {
      ProducerFactory<byte[], byte[]> pf =
          ((KafkaMessageChannelBinder) binders.getBinder(null, MessageChannel.class))
              .getTransactionalProducerFactory();
      KafkaTemplate<byte[], byte[]> template = new KafkaTemplate<>(requireNonNull(pf));
      container.setAfterRollbackProcessor(
          new DefaultAfterRollbackProcessor<>(
              new DeadLetterPublishingRecoverer(template), new FixedBackOff(2000L, 2L)));
    };
  }

  // Receive messages from `input`.
  // For each input, write an audit log to the database.
  // For each input, produce a message to both `left` and `right` atomically.
  // After three failed attempts to achieve the above, shuffle the message
  // off to `input.DLT` and move on.
  @Bean
  @Transactional
  public Consumer<String> persistAndSplit(
      StreamBridge bridge,
      AuditLogRepository repository
  ) {
    return input -> {
      bridge.send("left", ("left-" + input).getBytes());
      repository.createIfNotExists(input);

      if (input.startsWith("fail")) {
        throw new RuntimeException("Simulated error");
      }

      bridge.send("right", ("right-" + input).getBytes());
    };
  }

  @Bean
  public Consumer<Message<String>> logger() {
    return message -> {
      var receivedTopic = message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
      LOGGER.info("Received on topic=" + receivedTopic + " payload=" + message.getPayload());
    };
  }
}

spring:
  cloud:
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: 'tx-'
          required-acks: all
      bindings:
        persistAndSplit-in-0:
          destination: input
          group: input
        logger-in-0:
          destination: left,right,input.DLT
          group: logger
          consumer:
            properties:
              isolation.level: read_committed
    function:
      definition: persistAndSplit;logger

谢谢！

Answer 1

  @Bean
  @Transactional
  public Consumer<String> persistAndSplit(
      StreamBridge bridge,
      AuditLogRepository repository
  ) {

在本例中，@transactional位于bean定义中（在应用程序初始化期间只执行一次）；要获得运行时事务，您需要对lambda中的代码进行这样的注释；比如...

  @Bean
  public Consumer<String> persistAndSplit(
      StreamBridge bridge,
      AuditLogRepository repository,
      TxCode code
  ) {
    return Txcode:run;
  }

@Component
class TxCode {

    @Autowired
    AuditLogRepository repository

    @Autowired
    StreamBridge bridge;

    @Transactional
    void run(String input) {
      bridge.send("left", ("left-" + input).getBytes());
      repository.createIfNotExists(input);

      if (input.startsWith("fail")) {
        throw new RuntimeException("Simulated error");
      }

      bridge.send("right", ("right-" + input).getBytes());
    };
}

（或者你也可以通过桥并回购）。

return str -> code.run(str, repo, bridge);