@Transactional with RabbitTemplate.sendConnectionFactorySelectorExpression + SimpleRoutingConnectionFactory

[mychalvlcek]

Bug report

I am experiecing weird behavior when using SimpleRoutingConnectionFactory together with RabbitTemplate with filled sendConnectionFactorySelectorExpression. Then usage with @Transactional method, message has been sent immediately just like without any open transaction

var connectionFactory = new SimpleRoutingConnectionFactory();
// + setup some connections e.g. with key/lookup "someLookup"

RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory());
rabbitTemplate.setChannelTransacted(true);
var expression = exprParser.parseExpression("someLookup");
rabbitTemplate.setSendConnectionFactorySelectorExpression(expression);


@GetMapping("/test/rabbitmq")
@Transactional
public String rabbitTest() {
    rabbitTemplate.convertAndSend(MessagingConfig.EXCHANGE, MessagingConfig.RT, new MessageDto(1L, TenantContext.getTenantId()));
    System.out.println("breakpoint - in this point message is already sent");
    return "ok";
}

I am using this construct for publishing instead of SimpleResourceHolder.bind

Imo it is caused in RabbitTemplate, when creating channel - method obtainTargetConnectionFactory is used to obtain connection (this method takes into account sendConnectionFactorySelectorExpression) but then, when "transaction" context is evaluated, simple getter getConnectionFactory is used

[garyrussell]

Can you provide more information? Or, preferably, a MCRE.

I see obtainTargetConnectionFactory being called in send and the result passed to execute().

Then, in doExecute() it is that connection factory that is used to lookup the transactional resource.

Where are you seeing getConnectionFactory() being called?

This test passes:

@Test
@SuppressWarnings("unchecked")
public void testRoutingConnectionFactoryWithTx() throws Exception {
	org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory1 = Mockito
			.mock(org.springframework.amqp.rabbit.connection.ConnectionFactory.class);
	org.springframework.amqp.rabbit.connection.Connection connection1
			= mock(org.springframework.amqp.rabbit.connection.Connection.class);
	Channel channel1 = mock(Channel.class, "channel1");
	given(channel1.isOpen()).willReturn(true);
	willReturn(connection1).given(connectionFactory1).createConnection();
	given(connection1.isOpen()).willReturn(true);
	given(connection1.createChannel(true)).willReturn(channel1);

	org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory2 = Mockito
			.mock(org.springframework.amqp.rabbit.connection.ConnectionFactory.class);
	org.springframework.amqp.rabbit.connection.Connection connection2
			= mock(org.springframework.amqp.rabbit.connection.Connection.class);
	Channel channel2 = mock(Channel.class, "channel1");
	given(channel2.isOpen()).willReturn(true);
	willReturn(connection2).given(connectionFactory1).createConnection();
	given(connection2.isOpen()).willReturn(true);
	given(connection2.createChannel(true)).willReturn(channel2);

	Map<Object, org.springframework.amqp.rabbit.connection.ConnectionFactory> factories =
			new HashMap<Object, org.springframework.amqp.rabbit.connection.ConnectionFactory>(2);
	factories.put("foo", connectionFactory1);
	factories.put("bar", connectionFactory2);

	AbstractRoutingConnectionFactory connectionFactory = new SimpleRoutingConnectionFactory();
	connectionFactory.setTargetConnectionFactories(factories);
	connectionFactory.setDefaultTargetConnectionFactory(connectionFactory2);

	final RabbitTemplate template = new RabbitTemplate(connectionFactory);
	Expression expression = new SpelExpressionParser()
			.parseExpression("'foo'");
	template.setSendConnectionFactorySelectorExpression(expression);
	template.setChannelTransacted(true);

	new TransactionTemplate(new TestTransactionManager()).execute(status -> {
		template.convertAndSend("foo", "bar", "baz");
		return null;
	});

	Mockito.verify(connectionFactory1, times(1)).createConnection();
	Mockito.verify(connectionFactory2, never()).createConnection();
}

[garyrussell]

Aha - it does fail, if I use a RabbitTransactionManager...

		new TransactionTemplate(new RabbitTransactionManager(connectionFactory)).execute(status -> {
			template.convertAndSend("foo", "bar", "baz");
			return null;
		});

[mychalvlcek]

OK, thank you, or need any more detail?

[garyrussell]

Looking at it now... will let you know.

[garyrussell]

The issue is that when using the RabbitTransactionManager, the connection is made by the transaction manager, not the template, and he knows nothing about the lookup key.

Is there some reason that you must use a RabbitTransactionManager rather than just a local transaction, or transaction synchronization?

[mychalvlcek]

OK, hint from me, but am not really sure if I am missing something or not

RabbitTemplate.isChannelLocallyTransacted() looks like following:

protected boolean isChannelLocallyTransacted(Channel channel) {
	return isChannelTransacted() && !ConnectionFactoryUtils.isChannelTransactional(channel, getConnectionFactory());
}

and I think in that place it will not find correct object in some transaction holder object

[mychalvlcek]

The issue is that when using the RabbitTransactionManager, the connection is made by the transaction manager, not the template, and he knows nothing about the lookup key.

Is there some reason that you must use a RabbitTransactionManager rather than just a local transaction, or transaction synchronization?

I woulnt say I am using that, we using default/autoconfigured manager

[mychalvlcek]

I used sendConnectionFactorySelectorExpression to correctly "route" message instead of using suggested SimpleResourceHolder.bind

so edit our code to use SimpleResourceHolder.bind approach without need of sendConnectionFactorySelectorExpression and after those changes it works as expected

[garyrussell]

Yes, because then the transaction manager will get the right factory.

It's a limitation of using the expression with the template; it won't work if the TM creates the connection.

We should document that limitation - there is no way for the TM to know anything about the template's configuration.

That said, I don't see Boot auto-configuring a RabbitTransactionManager; if you have a small project that shows this behavior, I can take a look at it.

[garyrussell]

This code is correct:

return isChannelTransacted() && !ConnectionFactoryUtils.isChannelTransactional(channel, getConnectionFactory());

Because the key to the resource is the routing CF, not one of its delegates. If we looked up the delegate there, we would never find an existing transaction.