From d93a4a0c4a79513621dc617bc96838a50e45d6c6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 5 Oct 2022 13:22:16 -0400 Subject: [PATCH] GH-2510: RabbitMQ Super Streams Concurrency Add support for concurrency. Use try with resources in stream tests. Fix Test with Latest Boot Resolves #2524 --- .../RabbitExchangeQueueProvisioner.java | 7 +- .../rabbit/RabbitMessageChannelBinder.java | 2 +- .../stream/binder/rabbit/StreamUtils.java | 8 +- .../binder/rabbit/RabbitTestContainer.java | 2 +- .../stream/RabbitStreamBinderModuleTests.java | 118 ++++++++-------- ...bitStreamBinderModuleIntegrationTests.java | 129 ++++++++++++++++++ .../main/asciidoc/rabbit/rabbit_overview.adoc | 5 +- 7 files changed, 198 insertions(+), 73 deletions(-) create mode 100644 binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream2/RabbitStreamBinderModuleIntegrationTests.java diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index c93e87467..6c1d7bc2e 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -315,9 +315,10 @@ public class RabbitExchangeQueueProvisioner String routingKey = properties.getExtension().getBindingRoutingKey(); String rk = routingKey == null ? name : routingKey; - SuperStream ss = new SuperStream(name, properties.getInstanceCount(), (q, i) -> IntStream.range(0, i) - .mapToObj(j -> rk + "-" + j) - .collect(Collectors.toList())); + SuperStream ss = new SuperStream(name, properties.getInstanceCount() * properties.getConcurrency(), + (q, i) -> IntStream.range(0, i) + .mapToObj(j -> rk + "-" + j) + .collect(Collectors.toList())); synchronized (this.autoDeclareContext) { if (!this.autoDeclareContext.containsBean(name + ".superStream")) { this.autoDeclareContext.getBeanFactory().registerSingleton(name + ".superStream", ss); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 8db8fddb1..903613ca2 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -545,7 +545,7 @@ public class RabbitMessageChannelBinder extends RabbitConsumerProperties extension) { if (extension.getContainerType().equals(ContainerType.STREAM)) { - return StreamUtils.createContainer(consumerDestination, group, properties, destination, extension, + return StreamUtils.createContainer(consumerDestination, group, properties, destination, getApplicationContext()); } boolean directContainer = extension.getContainerType() diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java index 70a432509..8dd501dc5 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java @@ -32,6 +32,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerP import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.ProducerType; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler; @@ -72,7 +73,9 @@ public final class StreamUtils { */ public static MessageListenerContainer createContainer(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties properties, String destination, - RabbitConsumerProperties extension, AbstractApplicationContext applicationContext) { + ApplicationContext applicationContext) { + + RabbitConsumerProperties extension = properties.getExtension(); StreamListenerContainer container = new StreamListenerContainer(applicationContext.getBean(Environment.class)) { @@ -94,7 +97,8 @@ public final class StreamUtils { container.setStreamConverter(applicationContext.getBean(beanName, StreamMessageConverter.class)); } if (properties.getExtension().isSuperStream()) { - container.superStream(consumerDestination.getName(), consumerDestination.getName() + "." + group); + container.superStream(consumerDestination.getName(), consumerDestination.getName() + "." + group, + properties.getConcurrency()); } return container; } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java index 26df3f449..c7de2b242 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java @@ -29,7 +29,7 @@ public class RabbitTestContainer { private static final RabbitMQContainer RABBITMQ; static { - String image = "rabbitmq:management"; + String image = "rabbitmq:3.11-management"; String cache = System.getenv().get("IMAGE_CACHE"); if (cache != null) { image = cache + image; diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java index 5a7b6be7d..d4d2d9819 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java @@ -20,7 +20,6 @@ import com.rabbitmq.stream.ConsumerBuilder; import com.rabbitmq.stream.Environment; import com.rabbitmq.stream.OffsetSpecification; import com.rabbitmq.stream.ProducerBuilder; -import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.testcontainers.containers.RabbitMQContainer; @@ -29,7 +28,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.Binding; @@ -66,78 +64,71 @@ public class RabbitStreamBinderModuleTests { private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); - private ConfigurableApplicationContext context; - - @AfterEach - void tearDown() { - if (context != null) { - context.close(); - context = null; + @Test + void testStreamContainer() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class) + .web(WebApplicationType.NONE) + .run("--server.port=0")) { + BinderFactory binderFactory = context.getBean(BinderFactory.class); + RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, + MessageChannel.class); + RabbitConsumerProperties rProps = new RabbitConsumerProperties(); + rProps.setContainerType(ContainerType.STREAM); + ExtendedConsumerProperties props = + new ExtendedConsumerProperties(rProps); + props.setAutoStartup(false); + Binding binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props); + Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); + assertThat(container).isInstanceOf(StreamListenerContainer.class); + ((StreamListenerContainer) container).start(); + verify(context.getBean(ConsumerBuilder.class)).offset(OffsetSpecification.first()); + ((StreamListenerContainer) container).stop(); } } - @Test - void testStreamContainer() { - context = new SpringApplicationBuilder(SimpleProcessor.class) - .web(WebApplicationType.NONE) - .run("--server.port=0"); - BinderFactory binderFactory = context.getBean(BinderFactory.class); - RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, - MessageChannel.class); - RabbitConsumerProperties rProps = new RabbitConsumerProperties(); - rProps.setContainerType(ContainerType.STREAM); - ExtendedConsumerProperties props = - new ExtendedConsumerProperties(rProps); - props.setAutoStartup(false); - Binding binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props); - Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); - assertThat(container).isInstanceOf(StreamListenerContainer.class); - ((StreamListenerContainer) container).start(); - verify(this.context.getBean(ConsumerBuilder.class)).offset(OffsetSpecification.first()); - ((StreamListenerContainer) container).stop(); - } - @Test void testSuperStreamContainer() { - context = new SpringApplicationBuilder(SimpleProcessor.class) + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class) .web(WebApplicationType.NONE) - .run("--server.port=0"); - BinderFactory binderFactory = context.getBean(BinderFactory.class); - RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, - MessageChannel.class); - RabbitConsumerProperties rProps = new RabbitConsumerProperties(); - rProps.setContainerType(ContainerType.STREAM); - rProps.setSuperStream(true); - ExtendedConsumerProperties props = - new ExtendedConsumerProperties(rProps); - props.setAutoStartup(false); - props.setInstanceCount(1); - Binding binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props); - Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); - assertThat(container).isInstanceOf(StreamListenerContainer.class); - ((StreamListenerContainer) container).start(); - ConsumerBuilder builder = this.context.getBean(ConsumerBuilder.class); - verify(builder).singleActiveConsumer(); - verify(builder).superStream("testSuperStream"); - verify(builder).name("testSuperStream.grp"); - ((StreamListenerContainer) container).stop(); + .run("--server.port=0")) { + BinderFactory binderFactory = context.getBean(BinderFactory.class); + RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, + MessageChannel.class); + RabbitConsumerProperties rProps = new RabbitConsumerProperties(); + rProps.setContainerType(ContainerType.STREAM); + rProps.setSuperStream(true); + ExtendedConsumerProperties props = + new ExtendedConsumerProperties(rProps); + props.setAutoStartup(false); + props.setInstanceCount(1); + Binding binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props); + Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); + assertThat(container).isInstanceOf(StreamListenerContainer.class); + ((StreamListenerContainer) container).start(); + ConsumerBuilder builder = context.getBean(ConsumerBuilder.class); + verify(builder).singleActiveConsumer(); + verify(builder).superStream("testSuperStream"); + verify(builder).name("testSuperStream.grp"); + ((StreamListenerContainer) container).stop(); + } } @Test void testStreamHandler() { - context = new SpringApplicationBuilder(SimpleProcessor.class) + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class) .web(WebApplicationType.NONE) - .run("--server.port=0"); - BinderFactory binderFactory = context.getBean(BinderFactory.class); - RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, - MessageChannel.class); - RabbitProducerProperties rProps = new RabbitProducerProperties(); - rProps.setProducerType(ProducerType.STREAM_SYNC); - ExtendedProducerProperties props = - new ExtendedProducerProperties(rProps); - Binding binding = rabbitBinder.bindProducer("testStream", new DirectChannel(), props); - Object handler = TestUtils.getPropertyValue(binding, "val$producerMessageHandler"); - assertThat(handler).isInstanceOf(RabbitStreamMessageHandler.class); + .run("--server.port=0")) { + BinderFactory binderFactory = context.getBean(BinderFactory.class); + RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, + MessageChannel.class); + RabbitProducerProperties rProps = new RabbitProducerProperties(); + rProps.setProducerType(ProducerType.STREAM_SYNC); + ExtendedProducerProperties props = + new ExtendedProducerProperties(rProps); + Binding binding = rabbitBinder.bindProducer("testStream", new DirectChannel(), props); + Object handler = TestUtils.getPropertyValue(binding, "val$producerMessageHandler"); + assertThat(handler).isInstanceOf(RabbitStreamMessageHandler.class); + } } @SpringBootApplication(proxyBeanMethods = false) @@ -167,7 +158,6 @@ public class RabbitStreamBinderModuleTests { } @Bean - @ConditionalOnMissingBean Environment rabbitStreamEnvironment(ConsumerBuilder consumerBuilder, ProducerBuilder producerBuilder) { Environment env = mock(Environment.class); given(env.consumerBuilder()).willReturn(consumerBuilder); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream2/RabbitStreamBinderModuleIntegrationTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream2/RabbitStreamBinderModuleIntegrationTests.java new file mode 100644 index 000000000..a56e126de --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream2/RabbitStreamBinderModuleIntegrationTests.java @@ -0,0 +1,129 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.rabbit.stream2; + +import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.rabbitmq.stream.Address; +import com.rabbitmq.stream.OffsetSpecification; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; + +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.amqp.EnvironmentBuilderCustomizer; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Gary Russell + */ +public class RabbitStreamBinderModuleIntegrationTests { + + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); + + @Test + void testSuperStreamContainer() throws InterruptedException { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class) + .web(WebApplicationType.NONE) + .run("--server.port=0")) { + BinderFactory binderFactory = context.getBean(BinderFactory.class); + RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, + MessageChannel.class); + RabbitConsumerProperties rProps = new RabbitConsumerProperties(); + rProps.setContainerType(ContainerType.STREAM); + rProps.setSuperStream(true); + ExtendedConsumerProperties props = + new ExtendedConsumerProperties(rProps); + props.setAutoStartup(false); + props.setInstanceCount(3); + props.setConcurrency(3); + Binding binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props); + Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); + assertThat(container).isInstanceOf(StreamListenerContainer.class); + assertThat(container).extracting("concurrency").isEqualTo(3); + ((StreamListenerContainer) container).start(); + assertThat(context.getBean(SimpleProcessor.class).consumerCountLatch.await(10, TimeUnit.SECONDS)).isTrue(); + RabbitAdmin admin = context.getBean(RabbitAdmin.class); + System.out.println(RABBITMQ.getMappedPort(15672)); + for (int i = 0; i < 9; i++) { + Properties qProps = admin.getQueueProperties("testSuperStream-" + i); + assertThat(qProps).describedAs("Expected queue with index %d to exist", i).isNotNull(); + } + ((StreamListenerContainer) container).stop(); + } + } + + @SpringBootApplication(proxyBeanMethods = false) + public static class SimpleProcessor { + + final CountDownLatch consumerCountLatch = new CountDownLatch(3); + + @Bean + ConnectionFactory cf() { + return new CachingConnectionFactory(RABBITMQ.getMappedPort(5672)); + } + + @Bean + RabbitAdmin admin(ConnectionFactory cf) { + return new RabbitAdmin(cf); + } + + @Bean + EnvironmentBuilderCustomizer environmenCustomizer() { + return env -> env + .addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552))); + } + + @Bean + ListenerContainerCustomizer containerCustomizer() { + return (cont, dest, group) -> { + if (cont instanceof StreamListenerContainer container) { + container.setConsumerCustomizer((id, builder) -> { + builder.consumerUpdateListener(context -> { + this.consumerCountLatch.countDown(); + return OffsetSpecification.first(); + }); + }); + } + }; + } + + } + +} diff --git a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc index 3bcaa2854..a7c8b4e6a 100644 --- a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc +++ b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc @@ -420,7 +420,7 @@ Default: `1`. Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided. To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`. -IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is also not supported at this time. +IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is supported for super streams only. Only a single stream queue can be consumed by each binding. To configure the binder to use `containerType=stream`, Spring Boot will automatically configure an `Environment` `@Bean` from the application properties. @@ -493,12 +493,13 @@ public Consumer input() { spring.cloud.stream.bindings.input-in-0.destination=super spring.cloud.stream.bindings.input-in-0.group=test spring.cloud.stream.bindings.input-in-0.consumer.instance-count=3 +spring.cloud.stream.bindings.input-in-0.consumer.concurrency=3 spring.cloud.stream.rabbit.bindings.input-in-0.consumer.container-type=STREAM spring.cloud.stream.rabbit.bindings.input-in-0.consumer.super-stream=true ---- ==== -The framework will create a super stream named `super`, with 3 partitions. +The framework will create a super stream named `super`, with 9 partitions. Up to 3 instances of this application can be deployed. === Advanced Listener Container Configuration