From 060a67a716a975fce2ee09465bcf36f895dc840f Mon Sep 17 00:00:00 2001 From: buildmaster Date: Wed, 1 Dec 2021 16:39:57 +0000 Subject: [PATCH] Update SNAPSHOT to 3.2.1 --- README.adoc | 125 ++++++++++++++++-- docs/pom.xml | 2 +- docs/src/main/asciidoc/_configprops.adoc | 6 - pom.xml | 4 +- 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, 118 insertions(+), 27 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 b89646ffe..49089f526 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.1-SNAPSHOT + 3.2.1 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 14d55dc33..77a9e52a6 100644 --- a/pom.xml +++ b/pom.xml @@ -2,7 +2,7 @@ 4.0.0 spring-cloud-stream-binder-rabbit-parent - 3.2.1-SNAPSHOT + 3.2.1 pom org.springframework.cloud @@ -11,7 +11,7 @@ - 3.2.1-SNAPSHOT + 3.2.1 1.8 3.2.1 true diff --git a/spring-cloud-starter-stream-rabbit/pom.xml b/spring-cloud-starter-stream-rabbit/pom.xml index 94b0bf430..745839970 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.1-SNAPSHOT + 3.2.1 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 51a22700d..82aab0196 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.1-SNAPSHOT + 3.2.1 diff --git a/spring-cloud-stream-binder-rabbit-test-support/pom.xml b/spring-cloud-stream-binder-rabbit-test-support/pom.xml index 90fb4fe4b..0e094f774 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.1-SNAPSHOT + 3.2.1 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 b5e0f28fd..eed9a980e 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.1-SNAPSHOT + 3.2.1