Test binder doesn't support multiple bindings with the same group (message processed twice instead of once per group)

[jonenst]

Describe the issue
during unit tests with the test binder, with multiple bindings on the same group, a message sent once to the group is processed by all the bindings of the group, instead of by just one binding

To Reproduce
Use two bindings with the same group in one application. With the testbinder it processes the message twice instead of once. With the rabbitmq production binder it works: processed only once (just like when using a group with multiple replicas of the application)

package com.example.demo;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.function.Function;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.support.GenericMessage;

public class DemoApplicationTests {

	@Test
	public void sampleTest() {
		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
					TestChannelBinderConfiguration.getCompleteConfiguration(
							MyTestConfiguration.class))
					.run(
            "--spring.cloud.function.definition=uppercase1;uppercase2",
            "--spring.cloud.stream.bindings.uppercase1-in-0.destination=myInput",
            "--spring.cloud.stream.bindings.uppercase2-in-0.destination=myInput",
            "--spring.cloud.stream.bindings.uppercase1-out-0.destination=myOutput",
            "--spring.cloud.stream.bindings.uppercase2-out-0.destination=myOutput",
            "--spring.cloud.stream.bindings.uppercase1-in-0.group=mygroup",
            "--spring.cloud.stream.bindings.uppercase2-in-0.group=mygroup"
          )) {
			InputDestination source = context.getBean(InputDestination.class);
			OutputDestination target = context.getBean(OutputDestination.class);
			source.send(new GenericMessage<byte[]>("hello".getBytes()));
      var output1 = target.receive();
      var output2 = target.receive();
      System.out.println(output1);
      System.out.println(output2);
      assertThat(output2).isNull();
		}
	}

	@EnableAutoConfiguration
	public static class MyTestConfiguration {
		@Bean
		public Function<String, String> uppercase1() {
			return v -> {
        System.out.println("            >> WORK BEAN UPPERCASE1 <<");
        return v.toUpperCase();
      };
		}
		@Bean
		public Function<String, String> uppercase2() {
			return v -> {
        System.out.println("            >> WORK BEAN UPPERCASE2 <<");
        return v.toUpperCase();
      };
		}
	}
}

outputs

            >> WORK BEAN UPPERCASE1 <<
            >> WORK BEAN UPPERCASE2 <<
GenericMessage [payload=byte[5], headers={contentType=application/json, id=014d032e-93fb-113e-6dc4-9ee3b5875938, timestamp=1770894431724}]
GenericMessage [payload=byte[5], headers={contentType=application/json, id=5791fa33-8961-fc15-20cc-32f5d09329f8, timestamp=1770894431725}]
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 0.923 s <<< FAILURE! -- in com.example.demo.DemoApplicationTests

Version of the framework
<spring-cloud.version>2025.1.0</spring-cloud.version>
[INFO] +- org.springframework.cloud:spring-cloud-stream:jar:5.0.0:compile
[INFO] | +- org.springframework.cloud:spring-cloud-function-context:jar:5.0.0:compile
[INFO] | | +- org.springframework.cloud:spring-cloud-function-core:jar:5.0.0:compile
[INFO] - org.springframework.cloud:spring-cloud-stream-test-binder:jar:5.0.0:test

I have also seen it with
<spring-cloud.version>2024.0.2</spring-cloud.version>
[INFO] +- org.springframework.cloud:spring-cloud-stream:jar:4.2.2:compile
[INFO] | - org.springframework.cloud:spring-cloud-function-context:jar:4.2.3:compile
[INFO] | +- org.springframework.cloud:spring-cloud-function-core:jar:4.2.3:compile
[INFO] +- org.springframework.cloud:spring-cloud-stream-test-binder:jar:4.2.2:test

Expected behavior
A group consumes a message only once in tests just like in production by default
note: currently I workaround it by overriding the function definitions for tests to only keep 1 binding, but this makes my tests different than prod.
note2: I guess the test binder in addition to not supporting groups, doesn't rabbit x-priority, but should this be added as well so that I can unit test with the testbinder that my beans with lower priorities are only called when the beans with higher priorities are saturated ?

Thanks in advance !

Additional context
I use multiple bindings in the same app (maybe rare?) to be able to have different x-priority for each binding to spread the load of consuming messages accross all my replicas. (see

#3122 ,
spring-projects/spring-amqp#3092 ,

)

[olegz]

Test Bonder was never designed for handling groups due to the fact that groups are handled differently based on the messaging system. For example, Kafka provides native support for groups where with Rabbit we had to do some tricks tro provide groups support.
Test Binder was primarily designed to test message exchanges in unit tests. It was never designed to emulate full capabilities of various binders since those capabilities vary from binder to binder