diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index a9b060506..28de0a5f8 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -667,6 +667,7 @@ confirmAckChannel:: When `errorChannelEnabled` is true, a channel to which to send positive delivery acknowledgments (aka publisher confirms). If the channel does not exist, a `DirectChannel` is registered with this name. The connection factory must be configured to enable publisher confirms. +Mutually exclusive with `useConfirmHeader`. + Default: `nullChannel` (acks are discarded). deadLetterQueueName:: @@ -879,10 +880,131 @@ Default time (in milliseconds) to live to apply to the queue when declared. Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no limit` +useConfirmHeader:: +See <>. +Mutually exclusive with `confirmAckChannel`. ++ NOTE: In the case of RabbitMQ, content type headers can be set by external applications. Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport -- including transports, such as Kafka (prior to 0.11), that do not natively support headers. +[[publisher-confirms]] +=== Publisher Confirms + +There are two mechanisms to get the result of publishing a message; in each case, the connection factory must have `publisherConfirmType` set `ConfirmType.CORRELATED`. +The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see <>. + +The preferred mechanism, added in version 3.1 is to use a correlation data header and wait for the result via its `Future` property. +This is particularly useful with a batch listener because you can send multiple messages before waiting for the result. +To use this technique, set the `useConfirmHeader` property to true +The following simple application is an example of using this technique: + +==== +[source, properties] +---- +spring.cloud.stream.bindings.input-in-0.group=someGroup +spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true + +spring.cloud.stream.source=output +spring.cloud.stream.bindings.output-out-0.producer.error-channel-enabled=true + +spring.cloud.stream.rabbit.bindings.output-out-0.producer.useConfirmHeader=true +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.auto-bind-dlq=true +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10 + +spring.rabbitmq.publisher-confirm-type=correlated +spring.rabbitmq.publisher-returns=true +---- +==== + +==== +[source, java] +---- +@SpringBootApplication +public class Application { + + private static final Logger log = LoggerFactory.getLogger(Application.class); + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Autowired + private StreamBridge bridge; + + @Bean + Consumer> input() { + return list -> { + List results = new ArrayList<>(); + list.forEach(str -> { + log.info("Received: " + str); + MyCorrelationData corr = new MyCorrelationData(UUID.randomUUID().toString(), str); + results.add(corr); + this.bridge.send("output-out-0", MessageBuilder.withPayload(str.toUpperCase()) + .setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr) + .build()); + }); + results.forEach(correlation -> { + try { + Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS); + log.info(confirm + " for " + correlation.getPayload()); + if (correlation.getReturnedMessage() != null) { + log.error("Message for " + correlation.getPayload() + " was returned "); + + // try to re-publish, send a DLQ, etc + + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + e.printStackTrace(); + } + catch (ExecutionException | TimeoutException e) { + e.printStackTrace(); + } + }); + }; + } + + + @Bean + public ApplicationRunner runner(BatchingRabbitTemplate template) { + return args -> IntStream.range(0, 10).forEach(i -> + template.convertAndSend("input-in-0", "input-in-0.rbgh303", "foo" + i)); + } + + @Bean + public BatchingRabbitTemplate template(CachingConnectionFactory cf, TaskScheduler taskScheduler) { + BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 1000000, 1000); + return new BatchingRabbitTemplate(cf, batchingStrategy, taskScheduler); + } + +} + +class MyCorrelationData extends CorrelationData { + + private final String payload; + + MyCorrelationData(String id, String payload) { + super(id); + this.payload = payload; + } + + public String getPayload() { + return this.payload; + } + +} +---- +==== + +As you can see, we send each message and then await for the publication results. +If the messages can't be routed, then correlation data is populated with the returned message before the future is completed. + +IMPORTANT: The correlation data must be provided with a unique `id` so that the framework can perform the correlation. + +You cannot set both `useConfirmHeader` and `confirmAckChannel` but you can still receive returned messages in the error channel when `useConfirmHeader` is true, but using the correlation header is more convenient. + == Using Existing Queues/Exchanges By default, the binder will automatically provision a topic exchange with the name being derived from the value of the destination binding property ``. @@ -993,6 +1115,7 @@ RabbitMQ has two types of send failures: The latter is rare. According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.". +You can also get a negative acknowledgment if you publish to a bounded queue with `reject-publish` queue overflow behavior. As well as enabling producer error channels (as described in "`<>`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows: @@ -1013,6 +1136,8 @@ The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessa * `exchange`: The exchange to which the message was published. * `routingKey`: The routing key used when the message was published. +Also see <> for an alternative mechanism to receive returned messages. + For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties: * `failedMessage`: The spring-messaging `Message` that failed to be sent. diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index 4023ba612..629f8e2e3 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java @@ -92,6 +92,15 @@ public class RabbitProducerProperties extends RabbitCommonProperties { */ private String confirmAckChannel; + /** + * When true, the binding will complete the {@link java.util.concurrent.Future} field + * in a {@link org.springframework.amqp.rabbit.connection.CorrelationData} contained + * in the + * {@link org.springframework.amqp.support.AmqpHeaders#PUBLISH_CONFIRM_CORRELATION} + * header when the confirmation is received. + */ + private boolean useConfirmHeader; + /** * @deprecated - use {@link #setHeaderPatterns(String[])}. * @param requestHeaderPatterns the patterns. @@ -209,4 +218,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties { this.batchingStrategyBeanName = batchingStrategyBeanName; } + public boolean isUseConfirmHeader() { + return this.useConfirmHeader; + } + + public void setUseConfirmHeader(boolean useConfirmHeader) { + this.useConfirmHeader = useConfirmHeader; + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index d5974aea7..5f39495b3 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -300,11 +300,11 @@ public class RabbitMessageChannelBinder extends String exchangeName = producerDestination.getName(); String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length()); - final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint( - buildRabbitTemplate(producerProperties.getExtension(), - errorChannel != null)); - endpoint.setExchangeName(producerDestination.getName()); RabbitProducerProperties extendedProperties = producerProperties.getExtension(); + final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint( + buildRabbitTemplate(extendedProperties, + errorChannel != null || extendedProperties.isUseConfirmHeader())); + endpoint.setExchangeName(producerDestination.getName()); boolean expressionInterceptorNeeded = expressionInterceptorNeeded( extendedProperties); Expression routingKeyExpression = extendedProperties.getRoutingKeyExpression(); @@ -364,19 +364,25 @@ public class RabbitMessageChannelBinder extends if (errorChannel != null) { checkConnectionFactoryIsErrorCapable(); endpoint.setReturnChannel(errorChannel); - endpoint.setConfirmNackChannel(errorChannel); - String ackChannelBeanName = StringUtils - .hasText(extendedProperties.getConfirmAckChannel()) - ? extendedProperties.getConfirmAckChannel() - : IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME; - if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME) - && !getApplicationContext().containsBean(ackChannelBeanName)) { - GenericApplicationContext context = (GenericApplicationContext) getApplicationContext(); - context.registerBean(ackChannelBeanName, DirectChannel.class, - () -> new DirectChannel()); + if (!extendedProperties.isUseConfirmHeader()) { + endpoint.setConfirmNackChannel(errorChannel); + String ackChannelBeanName = StringUtils + .hasText(extendedProperties.getConfirmAckChannel()) + ? extendedProperties.getConfirmAckChannel() + : IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME; + if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME) + && !getApplicationContext().containsBean(ackChannelBeanName)) { + GenericApplicationContext context = (GenericApplicationContext) getApplicationContext(); + context.registerBean(ackChannelBeanName, DirectChannel.class, + () -> new DirectChannel()); + } + endpoint.setConfirmAckChannelName(ackChannelBeanName); + endpoint.setConfirmCorrelationExpressionString("#root"); + } + else { + Assert.state(!StringUtils.hasText(extendedProperties.getConfirmAckChannel()), + "You cannot specify a 'confirmAckChannel' when 'useConfirmHeader' is true"); } - endpoint.setConfirmAckChannelName(ackChannelBeanName); - endpoint.setConfirmCorrelationExpressionString("#root"); endpoint.setErrorMessageStrategy(new DefaultErrorMessageStrategy()); } endpoint.setHeadersMappedLast(true); @@ -861,8 +867,7 @@ public class RabbitMessageChannelBinder extends consumerProperties); } - private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, - boolean mandatory) { + private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) { RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { BatchingStrategy batchingStrategy = getBatchingStrategy(properties); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 8ec89f453..fa8e84397 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -64,6 +64,8 @@ import org.springframework.amqp.rabbit.batch.MessageBatch; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -245,7 +247,7 @@ public class RabbitBinderTests extends RabbitTestBinder binder = getBinder(); CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); ccf.setPublisherReturns(true); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); @@ -326,7 +328,7 @@ public class RabbitBinderTests extends RabbitTestBinder binder = getBinder(); CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); ccf.setPublisherReturns(true); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); @@ -350,6 +352,30 @@ public class RabbitBinderTests extends producerBinding.unbind(); } + @Test + public void testProducerConfirmHeader() throws Exception { + RabbitTestBinder binder = getBinder(); + CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + ccf.setPublisherReturns(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + ccf.resetConnection(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + ExtendedProducerProperties producerProps = createProducerProperties(); + producerProps.getExtension().setUseConfirmHeader(true); + Binding producerBinding = binder.bindProducer("confirms.0", + moduleOutputChannel, producerProps); + CorrelationData correlation = new CorrelationData("testConfirm"); + final Message message = MessageBuilder.withPayload("confirmsMessage".getBytes()) + .setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, correlation) + .build(); + moduleOutputChannel.send(message); + Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS); + assertThat(confirm.isAck()).isTrue(); + assertThat(correlation.getReturnedMessage()).isNotNull(); + producerBinding.unbind(); + } + @Test public void testConsumerProperties() throws Exception { RabbitTestBinder binder = getBinder();