Custom Webflux TransactionInterceptor gets invoked twice

I am using Spring Boot 3.4.5 + Webflux. I am building an integration with Hibernate Reactive. I want to support the timeout from @Transactional. The way to do that with Hibernate Reactive is to put a timeout on the flux/mono. I don't want users of my starter to have to do that manually, so I am attempting to use TransactionInterceptor.

public class CustomReactiveTransactionInterceptor extends TransactionInterceptor {

  public CustomReactiveTransactionInterceptor(
      TransactionManager txManager,
      TransactionAttributeSource attributeSource) {

    super(txManager, attributeSource);
  }

  @Override
  public Object invoke(MethodInvocation invocation) throws Throwable {
    // Determine the raw result of the method call
    Object result = invocation.proceed();

    System.out.println("CustomReactiveTransactionInterceptor invoked");

    Method method = invocation.getMethod();
    Class<?> targetClass = AopUtils.getTargetClass(invocation.getThis());

    TransactionAttribute txAttr =
        getTransactionAttributeSource().getTransactionAttribute(method, targetClass);

    // If it’s a Mono, wrap it in a reactive transaction
    if (result instanceof Mono<?> mono) {
      return TransactionalOperator
          .create((org.springframework.transaction.ReactiveTransactionManager) getTransactionManager())
          .transactional(mono);
    }

    // If it’s a Flux, similarly wrap
    if (result instanceof Flux<?> flux) {
      return TransactionalOperator
          .create((org.springframework.transaction.ReactiveTransactionManager) getTransactionManager())
          .transactional(flux/*.timeout(Duration.ofMillis(1))*/);
    }

    // Otherwise, fall back to the standard (imperative) behavior
    return super.invoke(invocation);
  }
}

And then I register it as:

  @Bean
  @Primary
  public CustomReactiveTransactionInterceptor txInterceptor(
      ReactiveTransactionManager reactiveTxManager,
      TransactionAttributeSource tas) {

    return new CustomReactiveTransactionInterceptor(reactiveTxManager, tas);
  }

  @Bean
  public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
      CustomReactiveTransactionInterceptor txInterceptor,
      TransactionAttributeSource tas) {

    var advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
    advisor.setTransactionAttributeSource(tas);
    advisor.setAdvice(txInterceptor);
    return advisor;
  }

And let's assume a controller method like this:

  @Transactional(timeout = 2)
  @GetMapping("1")
  public Flux<Film> test1() {
    return this.filmRepository
        .findAll()
        .thenMany(this.filmRepository.findAll(QFilm.film.title.startsWith("Ac")));
  }

The issue I'm running into is that the invoke method is triggered TWICE.

1. Flux.thenMany ⇢ at com.xxx.search2.test2.TestController.test1(TestController.java:38)
2. Flux.contextWrite ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.execute(TransactionalOperatorImpl.java:85)

Is that intended? My intention is to set the timeout on flux #1, but there doesn't seem to be a way to filter out the 2nd one.

For both invocations, the txAttribute is set and the method and targetClass are the same as well.

Start by tracing CustomReactiveTransactionInterceptor.invoke from TestController.test1 and compare the two paths shown in the stack traces, including Flux.thenMany and TransactionalOperatorImpl.execute. Read TransactionInterceptor and BeanFactoryTransactionAttributeSourceAdvisor to determine why the same method and transaction attributes are observed twice; completion requires establishing whether this invocation pattern is intended and identifying the relevant supported behavior.