Support service connections with RabbitMQ Streams and Testcontainers

[eddumelendez]

Add RabbitStreamConnectionDetails and support from RabbitMQContainer
when rabbitmq_stream plugin is enabled.

[mhalbritter]

Hey @eddumelendez, thanks for the PR. I've left some comments for your consideration.

[mhalbritter]

We don't use var in the codebase.

[wilkinsona]

when rabbitmq_stream plugin is enabled

Apologies if I've missed it, but I can't see any code in the connection details factory that checks that the stream plugin is enabled. What'll happen if the factory kicks in on a container without streams enabled? I'm wondering if it may break somehow, perhaps by overriding the app's Rabbit Streams configuration with connection details that won't work.

[eddumelendez]

rabbitmq_stream plugin uses a new port 5552. If the plugin is not enabled, then it will fail because can not get the mapped port. This also requires an additional bean, otherwise will fail. See https://www.rabbitmq.com/blog/2021/07/23/connecting-to-streams#using-the-stream-java-client-with-a-load-balancer

@Bean
EnvironmentBuilderCustomizer environmentBuilderCustomizer() {
	return env -> {
		Address entrypoint = new Address(rabbit.getHost(), rabbit.getMappedPort(RABBITMQ_STREAMS_PORT));
		env.addressResolver(address -> entrypoint);
	};
}

[eddumelendez]

PR has been updated

[snicoll]

Why do we need this? Especially given the javadoc, it looks like to me that the existing RabbitConnectionDetails should be enough, isn't it?

[eddumelendez]

rabbitmq stream uses a different port and in order to do that the plugin should be enabled. Having two implementations of RabbitConnectionDetails will lead to having an exception with 2 beans unless other beans or configuration are disabled when stream are enabled.

I think would be possible to have a different virtual host for stream in a single RabbitMQ instance, so, having a specific ConnectionDetails would make sense.

[wilkinsona]

I think there's a bug in the current code as we're not taking RabbitConnectionDetails into account when falling back from stream-specific configuration to general Rabbit configuration. I've opened #42489 for that. When fixed, we'll fallback from spring.rabbit.stream.* properties to RabbitConnectionDetails. With the changes here in place, we'd then fallback from RabbitStreamConnectionDetails (a replacement for some of the spring.rabbit.stream.* properties) to RabbitConnectionDetails (a replacement for some of the general spring.rabbit.* properties).

[wilkinsona]

Thinking about this some more, Rabbit Streams support is deceptively complicated (to me at least) as I think it's the first time where different users may want different things. One user may want RabbitConnectionDetails and RabbitStreamConnectionDetails to be sourced from the same container as their app is using the same Rabbit broker for both streaming and regular messaging. Another user may want RabbitConnectionDetails and RabbitStreamConnectionDetails to be sourced from different containers as their app is using different Rabbit brokers for streaming and regular messaging.

With only RabbitMQContainer to look at, even if we could detect in the container connection details factory that the streams plugin is enabled, we wouldn't know if the user wanted this container to be used only for streams or for both streams and regular messaging. It feels like we may need some other signal to indicate how the container should be used as a source for connection details. This reminds me a bit of the exploratory work being done for SSL and the possibility of having SSL-related annotations on the container field. I'll flag this one for a meeting so I can see what the rest of the team thinks.

[wilkinsona]

We think we may be able to check to see if the default streams port is exposed by the container. If it isn't, the factory should return null.

We should also think about Docker Compose as, ideally, we'd keep the supported services in sync across Compose and Testcontainers.

[wilkinsona]

Upon further thought, I'm not sure that this will work for all cases. Those cases are:

1. One container to be used for both streams and "standard" messaging
2. Two containers: one for streams and one for standard messaging

In the two container case, RabbitContainerConnectionDetailsFactory will match a container with the streams port exposed and I think we'll end up with a duplicate connection details failure. If we made RabbitContainerConnectionDetailsFactory ignore containers with the streams port exposed it would fix the two container case but it would break the one container case.

Some experimentation's needed to see if the above's accurate.

[wilkinsona]

I've rebased and polished the proposal in this branch.

We think we may be able to check to see if the default streams port is exposed by the container. If it isn't, the factory should return null.

This wasn't possible. The connection details factory can't access the container to examine its exposed port configuration as the container's only available to the connection details bean and one then once the bean has been initialized.

Upon further thought, I'm not sure that this will work for all cases. Those cases are:

1. One container to be used for both streams and "standard" messaging
2. Two containers: one for streams and one for standard messaging

In the two container case, RabbitContainerConnectionDetailsFactory will match a container with the streams port exposed and I think we'll end up with a duplicate connection details failure. If we made RabbitContainerConnectionDetailsFactory ignore containers with the streams port exposed it would fix the two container case but it would break the one container case.

Some experimentation's needed to see if the above's accurate.

I've done some experimentation and we don't end up with a duplicate connection details failure, instead any connection details of the same type after the first are ignored:

spring-boot/core/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/ConnectionDetailsRegistrar.java

Lines 98 to 103 in 1cf25e9

	String[] existingBeans = this.beanFactory.getBeanNamesForType(connectionDetailsType);
	if (!ObjectUtils.isEmpty(existingBeans)) {
	logger.debug(LogMessage.of(() -> "Skipping registration of %s due to existing beans %s".formatted(source,
	Arrays.asList(existingBeans))));
	return;
	}

Given the two behaviors described above, the polished proposal requires the use of the type attribute on @ServiceConnection to opt into streams support:

• With no type attribute, only a RabbitConnectionDetails will the created
• With type = RabbitStreamConnectionDetails.class, only a RabbitStreamConnectionDetails will be created
• With type = { RabbitConnectionDetails.class, RabbitStreamConnectionDetails.class }, both RabbitConnectionDetails and RabbitStreamConnectionDetails will be created.

In the two cases where RabbitStreamConnectionDetails is created, port 5552 must be exposed by the container otherwise a failure will occur when the connection details implementation tries to get the mapped port.

This opt-in behavior and the need for the type attribute is new and, for now at least, unique to RabbitStreamConnectionDetails. I'd like to discuss the approach with the rest of the team and see if we're happy with this direction, particularly as it requires a small change to the public API of ContainerConnectionSource to all the connection details factory to access the requested connection details types.

If we are happy, the next step will be to consider Docker Compose support and how things might work there.

[wilkinsona]

We discussed this today and wondered if we could overload sourceAccepts rather than make the source's accessor public.

[wilkinsona]

We discussed this today and wondered if we could overload sourceAccepts rather than make the source's accessor public.

I don't think this really helps. You either have to implement some logic in ContainerConnectionDetailsFactory.sourceAccepts(ContainerConnectionSource<C>, Class<?>, Class<?>) or ContainerConnectionSource.accepts(String, Class<?>, Class<?>) that it calls needs to be overloaded.

The former leaves a split in the code where checking if a source accepts a particular connection is split across the factory and the source. The latter is public API so overloading it doesn't avoid an increase in the public API. I think making getConnectionDetailsTypes() public is the cleanest way to do this.

[wilkinsona]

Thanks very much, @eddumelendez.