Make TransactionalApplicationListenerAdapter support fallbackExecution

[reda-alaoui]

I have the following use case.

I have a bean that should act as a multi node event router.
The bean must listen for a particular event interface and broadcast a received event to the other nodes. If a transaction is running, event should be broadcast after commit. If there is no transaction, the event should be immediatly broadcast.

The event interface looks like this:

interface MultiNodeClusterEvent extends Serializable {}

The router looks like this:

/**
 * @author Réda Housni Alaoui
 */
@Component
class MultiNodeClusterEventRouter {

  private final ApplicationEventPublisher eventPublisher;

  MultiNodeClusterEventRouter(ApplicationEventPublisher eventPublisher) {
    this.eventPublisher = requireNonNull(eventPublisher);
  }

  @TransactionalEventListener(fallbackExecution = true)
  public void onLocalEvent(MultiNodeClusterEvent event) {
    // Broadcast to other nodes
  }

  // Triggered by JGroup
  private void onRemoteEvent(MultiNodeClusterEvent event) {
    eventPublisher.publishEvent(event);
  }

}

The obvious issue with this code is that the routers (plural because 1 router/node) may enter in an infinite broadcast loop. The easiest solution that comes to mind is to make the router check the event source and ignore the event if the source is itself. But since I am listening for a raw event, I don't have access to ApplicationEvent informations.

I know that under the cover, Spring use PayloadApplicationEvent for raw events. So I try to implement the infinite loop guard like this:

/**
 * @author Réda Housni Alaoui
 */
@Component
class MultiNodeClusterEventRouter extends TransactionalApplicationListenerAdapter<PayloadApplicationEvent<MultiNodeClusterEvent>> {

  private final ApplicationEventPublisher eventPublisher;

  MultiNodeClusterEventRouter(ApplicationEventPublisher eventPublisher) {
    super(event -> {
      // Missing fallbackExecution !
      // If event.source == this, ignore.
      // Else, broadcast to other nodes
   });
    this.eventPublisher = requireNonNull(eventPublisher);
  }

  // Triggered by JGroup
  private void onRemoteEvent(MultiNodeClusterEvent event) {
    eventPublisher.publishEvent(new PayloadApplicationEvent<>(event));
  }

}

But when I do this, I loose TransactionalApplicationListenerMethodAdapter#fallbackExecution feature.

Could we add this feature to TransactionalApplicationListenerAdapter and pass its activation as an argument to TransactionalApplicationListener#forPayload ?

[jhoeller]

Would it also work if you declare your @TransactionalEventListener method as onLocalEvent(PayloadApplicationEvent<MultiNodeClusterEvent>), for the purpose of checking the event source?

Also, would you actually use the exposure of that flag on TransactionalApplicationListener#forPayload? If we need to add that capability there, I'd rather just put it on the TransactionalApplicationListenerAdapter class.

[reda-alaoui]

Would it also work if you declare your @TransactionalEventListener method as onLocalEvent(PayloadApplicationEvent), for the purpose of checking the event source?

I tried this way before creating this issue, but the listener method is never called. Is that supposed to work? IMO, that would be the most elegant way of handling this use case.

Also, would you actually use the exposure of that flag on TransactionalApplicationListener#forPayload? If we need to add that capability there, I'd rather just put it on the TransactionalApplicationListenerAdapter class.

Adding the capability only to TransactionalApplicationListenerAdapter would be more than enough for my use case. Do you want me to modify #34681 accordingly ?

[jhoeller]

Let's try to find out why onLocalEvent(PayloadApplicationEvent<MultiNodeClusterEvent>) did not work for you. This should be working, we also got tests for this around @TransactionalEventListener specifically. Ideally we would not need to add the fallback capability to TransactionalApplicationListenerAdapter at all then.

If MultiNodeClusterEvent is an interface and the payload is being published with the concrete type (that depends a bit on how it is being published), you might need to declare onLocalEvent(PayloadApplicationEvent<? extends MultiNodeClusterEvent>) for correct use of generics (due to the strict rules for nested generic type matching). Could you double-check this?