From 8aef57712d8d261849cd836492cf93a7013a9cf4 Mon Sep 17 00:00:00 2001 From: buildmaster Date: Wed, 1 Dec 2021 12:57:49 +0000 Subject: [PATCH] Update SNAPSHOT to 3.2.0 --- README.adoc | 125 ++++++++++++++++-- docs/pom.xml | 2 +- docs/src/main/asciidoc/_configprops.adoc | 6 - pom.xml | 8 +- spring-cloud-starter-stream-rabbit/pom.xml | 2 +- .../pom.xml | 2 +- .../pom.xml | 2 +- spring-cloud-stream-binder-rabbit/pom.xml | 2 +- 8 files changed, 120 insertions(+), 29 deletions(-) diff --git a/README.adoc b/README.adoc index 8bf85c9d5..991ab0f4c 100644 --- a/README.adoc +++ b/README.adoc @@ -116,7 +116,7 @@ A comma-separated list of RabbitMQ management plugin URLs. Only used when `nodes` contains more than one entry. Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`. Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue. -See https://docs.spring.io/spring-amqp/reference/html/#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. +See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. + Default: empty. spring.cloud.stream.rabbit.binder.nodes:: @@ -185,6 +185,7 @@ Default: none - the broker will generate random consumer tags. containerType:: Select the type of listener container to be used. See https://docs.spring.io/spring-amqp/reference/html/_reference.html#choose-container[Choosing a Container] in the Spring AMQP documentation for more information. +Also see <>. + Default: `simple` deadLetterQueueName:: @@ -434,6 +435,65 @@ Not supported when the `containerType` is `direct`. + Default: `1`. +[[rabbitmq-stream-consumer]] +=== Initial Consumer Support for the RabbitMQ Stream Plugin + +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. +Only a single stream queue can be consumed by each binding. + +To configure the binder to use `containerType=stream`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the listener container. + +==== +[source, java] +---- +@Bean +Environment streamEnv() { + return Environment.builder() + .build(); +} + +@Bean +ListenerContainerCustomizer customizer() { + return (cont, dest, group) -> { + StreamListenerContainer container = (StreamListenerContainer) cont; + container.setConsumerCustomizer((name, builder) -> { + builder.offset(OffsetSpecification.first()); + }); + // ... + }; +} +---- +==== + +The `name` argument passed to the customizer is `destination + '.' + group + '.container'`. + +The stream `name()` (for the purpose of offset tracking) is set to the binding `destination + '.' + group`. +It can be changed using a `ConsumerCustomizer` shown above. +If you decide to use manual offset tracking, the `Context` is available as a message header: + +==== +[source, java] +---- +int count; + +@Bean +public Consumer> input() { + return msg -> { + System.out.println(msg); + if (++count % 1000 == 0) { + Context context = msg.getHeaders().get("rabbitmq_streamContext", Context.class); + context.consumer().store(context.offset()); + } + }; +} +---- +==== + +Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and consumer builder. + === Advanced Listener Container Configuration To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context. @@ -977,16 +1037,16 @@ public class Application { if (correlation.getReturnedMessage() != null) { log.error("Message for " + correlation.getPayload() + " was returned "); - // try to re-publish, send a DLQ, etc + // throw some exception to invoke binder retry/error handling } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - e.printStackTrace(); + throw new IllegalStateException(e); } catch (ExecutionException | TimeoutException e) { - e.printStackTrace(); + throw new IllegalStateException(e); } }); }; @@ -1044,7 +1104,7 @@ There are a number of rabbit-specific binding properties that allow you to modif If you have an existing exchange/queue that you wish to use, you can completely disable automatic provisioning as follows, assuming the exchange is named `myExchange` and the queue is named `myQueue`: -* `spring.cloud.stream.bindings..destination=myExhange` +* `spring.cloud.stream.bindings..destination=myExchange` * `spring.cloud.stream.bindings..group=myQueue` * `spring.cloud.stream.rabbit.bindings..consumer.bindQueue=false` * `spring.cloud.stream.rabbit.bindings..consumer.declareExchange=false` @@ -1172,6 +1232,41 @@ For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageEx There is no automatic handling of these exceptions (such as sending to a <>). You can consume these exceptions with your own Spring Integration flow. +[[rabbitmq-stream-producer]] +=== Initial Producer Support for the RabbitMQ Stream Plugin + +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 producer properties described above are not supported when you set the `producerType` property to `STREAM_SYNC` or `STREAM_ASYNC`. + +To configure the binder to use a stream `ProducerType`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the message handler. + +==== +[source, java] +---- +@Bean +Environment streamEnv() { + return Environment.builder() + .build(); +} + +@Bean +ProducerMessageHandlerCustomizer handlerCustomizer() { + return (hand, dest) -> { + RabbitStreamMessageHandler handler = (RabbitStreamMessageHandler) hand; + handler.setConfirmTimeout(5000); + ((RabbitStreamTemplate) handler.getStreamOperations()).setProducerCustomizer( + (name, builder) -> { + ... + }); + }; +} +---- +==== + +Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and producer builder. + = Appendices [appendix] [[building]] @@ -1181,22 +1276,19 @@ You can consume these exceptions with your own Spring Integration flow. === Basic Compile and Test -Pre-requisites: - -* To compile, JDK {jdkversion} installed. -* To run tests, RabbitMQ server running on `localhost:5672` - +To build the source you will need to install JDK {jdkversion}. The build uses the Maven wrapper so you don't have to install a specific -version of Maven. The main build command is +version of Maven. To enable the tests, you should have RabbitMQ server running +on localhost and the default port (5672) +before building. + +The main build command is ---- $ ./mvnw clean install ---- -NOTE: There are scripts in `./ci-docker-compose` that use https://docs.docker.com/compose//[Docker Compose] to -start/stop a local RabbitMQ server. - You can also add '-DskipTests' if you like, to avoid running the tests. NOTE: You can also install Maven (>=3.3.3) yourself and run the `mvn` command @@ -1212,6 +1304,11 @@ build succeed, please raise a ticket to get the settings added to source control. +The projects that require middleware generally include a +`docker-compose.yml`, so consider using +https://compose.docker.io/[Docker Compose] to run the middeware servers +in Docker containers. + === Documentation There is a "docs" profile that will generate documentation. diff --git a/docs/pom.xml b/docs/pom.xml index 716e65be1..6f05dedc2 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0 jar spring-cloud-stream-binder-rabbit-docs diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 88ac8ca3b..6b575783b 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -9,7 +9,6 @@ |spring.cloud.stream.dynamic-destinations | `[]` | A list of destinations that can be bound dynamically. If set, only listed destinations can be bound. |spring.cloud.stream.function.batch-mode | `false` | |spring.cloud.stream.function.bindings | | -|spring.cloud.stream.function.definition | | Definition of functions to bind. If several functions need to be composed into one, use pipes (e.g., 'fooFunc\|barFunc') |spring.cloud.stream.instance-count | `1` | The number of deployed instances of an application. Default: 1. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-count" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index | `0` | The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index-list | | A list of instance id's from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is the name of the binding. This setting will override the one set in 'spring.cloud.stream.instance-index' @@ -21,11 +20,6 @@ |spring.cloud.stream.metrics.schedule-interval | `60s` | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds |spring.cloud.stream.override-cloud-connectors | `false` | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems. |spring.cloud.stream.pollable-source | `none` | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding -|spring.cloud.stream.poller.cron | | Cron expression value for the Cron Trigger. -|spring.cloud.stream.poller.fixed-delay | `1000` | Fixed delay for default poller. -|spring.cloud.stream.poller.initial-delay | `0` | Initial delay for periodic triggers. -|spring.cloud.stream.poller.max-messages-per-poll | `1` | Maximum messages per poll for the default poller. -|spring.cloud.stream.poller.time-unit | | The TimeUnit to apply to delay values. |spring.cloud.stream.rabbit.binder.admin-addresses | `[]` | Urls for management plugins; only needed for queue affinity. |spring.cloud.stream.rabbit.binder.admin-adresses | | |spring.cloud.stream.rabbit.binder.compression-level | `0` | Compression level for compressed bindings; see 'java.util.zip.Deflator'. diff --git a/pom.xml b/pom.xml index 0610fb948..07243417b 100644 --- a/pom.xml +++ b/pom.xml @@ -2,18 +2,18 @@ 4.0.0 spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0 pom org.springframework.cloud spring-cloud-build - 3.1.0-SNAPSHOT + 3.1.0 - 3.2.0-SNAPSHOT + 3.2.0 1.8 - 3.2.0-SNAPSHOT + 3.2.0 true true diff --git a/spring-cloud-starter-stream-rabbit/pom.xml b/spring-cloud-starter-stream-rabbit/pom.xml index a8fe1082d..ac5ef7656 100644 --- a/spring-cloud-starter-stream-rabbit/pom.xml +++ b/spring-cloud-starter-stream-rabbit/pom.xml @@ -4,7 +4,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0 spring-cloud-starter-stream-rabbit Spring Cloud Starter Stream Rabbit diff --git a/spring-cloud-stream-binder-rabbit-core/pom.xml b/spring-cloud-stream-binder-rabbit-core/pom.xml index ee954e02a..5288d6f16 100644 --- a/spring-cloud-stream-binder-rabbit-core/pom.xml +++ b/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0 diff --git a/spring-cloud-stream-binder-rabbit-test-support/pom.xml b/spring-cloud-stream-binder-rabbit-test-support/pom.xml index ed10c0e8b..0f44a2dc0 100644 --- a/spring-cloud-stream-binder-rabbit-test-support/pom.xml +++ b/spring-cloud-stream-binder-rabbit-test-support/pom.xml @@ -4,7 +4,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0 spring-cloud-stream-binder-rabbit-test-support Rabbit related test classes diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 6e17200c1..c14841e22 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0