diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecoverer.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecoverer.java new file mode 100644 index 000000000..628ca6c0f --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecoverer.java @@ -0,0 +1,52 @@ +package org.springframework.cloud.stream.binder.rabbit; + +import java.util.List; +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer; +import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; +import org.springframework.util.Assert; + +public class BatchCapableRejectAndDontRequeueRecoverer implements MessageBatchRecoverer { + + protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR protected + + private final Supplier messageSupplier; + + public BatchCapableRejectAndDontRequeueRecoverer() { + this(() -> "Retry Policy Exhausted"); + } + + /** + * Construct an instance with the provided exception message supplier. + * + * @param messageSupplier the message supplier. + */ + public BatchCapableRejectAndDontRequeueRecoverer(Supplier messageSupplier) { + Assert.notNull(messageSupplier, "'messageSupplier' cannot be null"); + this.messageSupplier = messageSupplier; + } + + @Override + public void recover(List messages, Throwable cause) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("Retries exhausted for message " + messages, cause); + } + throw new ListenerExecutionFailedException(this.messageSupplier.get(), + new AmqpRejectAndDontRequeueException(cause), messages.toArray(Message[]::new)); + } + + @Override + public void recover(Message message, Throwable cause) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("Retries exhausted for message " + message, cause); + } + throw new ListenerExecutionFailedException(this.messageSupplier.get(), + new AmqpRejectAndDontRequeueException(cause), message); + } +} 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 903613ca2..ea430cce3 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 @@ -32,6 +32,7 @@ import java.util.regex.Pattern; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Envelope; +import org.jetbrains.annotations.NotNull; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; @@ -54,7 +55,6 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; -import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; @@ -121,8 +121,7 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** - * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by - * RabbitMQ. + * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ. * * @author Mark Fisher * @author Gary Russell @@ -209,8 +208,9 @@ public class RabbitMessageChannelBinder extends } /** - * Set a {@link MessagePostProcessor} to decompress messages. Defaults to a - * {@link DelegatingDecompressingPostProcessor} with its default delegates. + * Set a {@link MessagePostProcessor} to decompress messages. Defaults to a {@link + * DelegatingDecompressingPostProcessor} with its default delegates. + * * @param decompressingPostProcessor the post processor. */ public void setDecompressingPostProcessor( @@ -219,9 +219,9 @@ public class RabbitMessageChannelBinder extends } /** - * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress - * messages. Defaults to a - * {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}. + * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages. Defaults to a {@link + * org.springframework.amqp.support.postprocessor.GZipPostProcessor}. + * * @param compressingPostProcessor the post processor. */ public void setCompressingPostProcessor( @@ -279,8 +279,7 @@ public class RabbitMessageChannelBinder extends } /** - * Get the underlying {@link ConnectionFactory} instance to - * allow manually altering the connection lifecycle. + * Get the underlying {@link ConnectionFactory} instance to allow manually altering the connection lifecycle. */ public ConnectionFactory getConnectionFactory() { return this.connectionFactory; @@ -398,8 +397,8 @@ public class RabbitMessageChannelBinder extends endpoint.setConfirmNackChannel(errorChannel); String ackChannelBeanName = StringUtils .hasText(extendedProperties.getConfirmAckChannel()) - ? extendedProperties.getConfirmAckChannel() - : IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME; + ? extendedProperties.getConfirmAckChannel() + : IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME; if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME) && !getApplicationContext().containsBean(ackChannelBeanName)) { GenericApplicationContext context = (GenericApplicationContext) getApplicationContext(); @@ -453,7 +452,7 @@ public class RabbitMessageChannelBinder extends Expression delayExpression = extendedProperties.getDelayExpression(); return (rkExpression != null && interceptorNeededPattern.matcher(rkExpression.getExpressionString()).find()) || (delayExpression != null - && interceptorNeededPattern.matcher(delayExpression.getExpressionString()).find()); + && interceptorNeededPattern.matcher(delayExpression.getExpressionString()).find()); } private void checkConnectionFactoryIsErrorCapable() { @@ -486,9 +485,9 @@ public class RabbitMessageChannelBinder extends boolean rootIsExpression) { String partitionRoutingExpression = rootIsExpression ? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER - + "']" + + "']" : "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER - + "']"; + + "']"; return new SpelExpressionParser().parseExpression(partitionRoutingExpression); } @@ -692,6 +691,7 @@ public class RabbitMessageChannelBinder extends private final ConfirmType confirmType; + { this.template.setUsePublisherConnection(true); this.template.setChannelTransacted(properties.getExtension().isTransacted()); @@ -721,13 +721,13 @@ public class RabbitMessageChannelBinder extends @Override public void handleMessage(org.springframework.messaging.Message message) throws MessagingException { - Message amqpMessage = StaticMessageHeaderAccessor.getSourceData(message); + List amqpMessages = extractAmqpMessages(message, properties); if (!(message instanceof ErrorMessage)) { logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: " + message); } - else if (amqpMessage == null) { + else if (amqpMessages == null) { logger.error("No raw message header in " + message); } else { @@ -741,9 +741,7 @@ public class RabbitMessageChannelBinder extends } return; } - MessageProperties messageProperties = amqpMessage - .getMessageProperties(); - Map headers = messageProperties.getHeaders(); + String stackTraceAsString = getStackTraceAsString(cause); if (this.maxStackTraceLength < 0) { int rabbitMaxStackTraceLength = RabbitUtils @@ -763,23 +761,15 @@ public class RabbitMessageChannelBinder extends + "consider increasing frame_max on the broker or reduce the stack trace depth", cause); } - headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, - stackTraceAsString); - headers.put(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE, - cause.getCause() != null ? cause.getCause().getMessage() - : cause.getMessage()); - headers.put(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE, - messageProperties.getReceivedExchange()); - headers.put(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY, - messageProperties.getReceivedRoutingKey()); - if (properties.getExtension().getRepublishDeliveyMode() != null) { - messageProperties.setDeliveryMode( - properties.getExtension().getRepublishDeliveyMode()); + for (Message amqpMessage : amqpMessages) { + MessageProperties messageProperties = adjustMessagePropertiesHeader(cause, + stackTraceAsString, + amqpMessage); + doSend(this.exchange, + this.routingKey != null ? this.routingKey + : messageProperties.getConsumerQueue(), + amqpMessage); } - doSend(this.exchange, - this.routingKey != null ? this.routingKey - : messageProperties.getConsumerQueue(), - amqpMessage); if (properties.getExtension().getAcknowledgeMode().equals(AcknowledgeMode.MANUAL)) { org.springframework.messaging.Message original = ((ErrorMessage) message).getOriginalMessage(); @@ -787,8 +777,8 @@ public class RabbitMessageChannelBinder extends // If we are using manual acks, ack the original message. try { original.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class) - .basicAck(original.getHeaders() - .get(AmqpHeaders.DELIVERY_TAG, Long.class), false); + .basicAck(original.getHeaders() + .get(AmqpHeaders.DELIVERY_TAG, Long.class), false); } catch (IOException e) { logger.debug("Failed to ack original message", e); @@ -798,6 +788,27 @@ public class RabbitMessageChannelBinder extends } } + @NotNull + private MessageProperties adjustMessagePropertiesHeader(Throwable cause, String stackTraceAsString, Message amqpMessage) { + MessageProperties messageProperties = amqpMessage + .getMessageProperties(); + Map headers = messageProperties.getHeaders(); + headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, + stackTraceAsString); + headers.put(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE, + cause.getCause() != null ? cause.getCause().getMessage() + : cause.getMessage()); + headers.put(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE, + messageProperties.getReceivedExchange()); + headers.put(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY, + messageProperties.getReceivedRoutingKey()); + if (properties.getExtension().getRepublishDeliveyMode() != null) { + messageProperties.setDeliveryMode( + properties.getExtension().getRepublishDeliveyMode()); + } + return messageProperties; + } + private void doSend(String exchange, String routingKey, Message amqpMessage) { if (ConfirmType.SIMPLE.equals(this.confirmType)) { this.template.invoke(temp -> { @@ -877,18 +888,17 @@ public class RabbitMessageChannelBinder extends } return !(cause instanceof ImmediateAcknowledgeAmqpException); } - }; } else if (properties.getMaxAttempts() > 1) { return new MessageHandler() { - private final RejectAndDontRequeueRecoverer recoverer = new RejectAndDontRequeueRecoverer(); + private final BatchCapableRejectAndDontRequeueRecoverer recoverer = new BatchCapableRejectAndDontRequeueRecoverer(); @Override public void handleMessage( org.springframework.messaging.Message message) throws MessagingException { - Message amqpMessage = StaticMessageHeaderAccessor.getSourceData(message); + List amqpMessages = extractAmqpMessages(message, properties); /* * NOTE: The following IF and subsequent ELSE IF should never happen * under normal interaction and it should always go to the last ELSE @@ -904,14 +914,14 @@ public class RabbitMessageChannelBinder extends "Unexpected error message " + message, new AmqpRejectAndDontRequeueException(""), (Message[]) null); } - else if (amqpMessage == null) { + else if (amqpMessages == null || amqpMessages.isEmpty()) { logger.error("No raw message header in " + message); throw new ListenerExecutionFailedException( "Unexpected error message " + message, - new AmqpRejectAndDontRequeueException(""), amqpMessage); + new AmqpRejectAndDontRequeueException(""), amqpMessages.toArray(Message[]::new)); } else { - this.recoverer.recover(amqpMessage, + this.recoverer.recover(amqpMessages, (Throwable) message.getPayload()); } } @@ -923,6 +933,18 @@ public class RabbitMessageChannelBinder extends } } + private List extractAmqpMessages(org.springframework.messaging.Message message, ExtendedConsumerProperties properties) { + + if (properties.isBatchMode() || properties.getExtension().isEnableBatching()) { + logger.debug("Batch mode enabled: Extract list instead of single message"); + return StaticMessageHeaderAccessor.getSourceData(message); + } + else { + Message amqpMessage = StaticMessageHeaderAccessor.getSourceData(message); + return List.of(amqpMessage); + } + } + @Override protected MessageHandler getPolledConsumerErrorMessageHandler( ConsumerDestination destination, String group, @@ -964,11 +986,11 @@ public class RabbitMessageChannelBinder extends }; } -// @Override -// protected String errorsBaseName(ConsumerDestination destination, String group, -// ExtendedConsumerProperties consumerProperties) { -// return destination.getName() + ".errors"; -// } + // @Override + // protected String errorsBaseName(ConsumerDestination destination, String group, + // ExtendedConsumerProperties consumerProperties) { + // return destination.getName() + ".errors"; + // } private String deadLetterExchangeName(RabbitCommonProperties properties) { if (properties.getDeadLetterExchange() == null) { diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecovererTest.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecovererTest.java new file mode 100644 index 000000000..1490b8b1f --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecovererTest.java @@ -0,0 +1,56 @@ +package org.springframework.cloud.stream.binder.rabbit; + +import java.util.List; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer; +import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class BatchCapableRejectAndDontRequeueRecovererTest { + + @Test + void testBatchRecoverThrowExceptionCorrectly() { + MessageBatchRecoverer messageBatchRecoverer = new BatchCapableRejectAndDontRequeueRecoverer(); + + Message expectedMessage1 = new Message(new byte[] {}); + Message expectedMessage2 = new Message(new byte[] {}); + + List messages = List.of(expectedMessage1, expectedMessage2); + + Throwable expectedThrowable = new RuntimeException("test"); + + ListenerExecutionFailedException exception = assertThrows(ListenerExecutionFailedException.class, + () -> messageBatchRecoverer.recover(messages, expectedThrowable)); + + assertTrue(exception.getFailedMessages().contains(expectedMessage1)); + assertTrue(exception.getFailedMessages().contains(expectedMessage2)); + assertTrue(exception.getCause() instanceof AmqpRejectAndDontRequeueException); + assertEquals(expectedThrowable, exception.getCause().getCause()); + + } + + @Test + void testRecoverThrowExceptionCorrectly() { + MessageBatchRecoverer messageBatchRecoverer = new BatchCapableRejectAndDontRequeueRecoverer(); + + Message expectedMessage = new Message(new byte[] {}); + + Throwable expectedThrowable = new RuntimeException("test"); + + ListenerExecutionFailedException exception = assertThrows(ListenerExecutionFailedException.class, + () -> messageBatchRecoverer.recover(expectedMessage, expectedThrowable)); + + assertEquals(expectedMessage, exception.getFailedMessage()); + assertTrue(exception.getCause() instanceof AmqpRejectAndDontRequeueException); + assertEquals(expectedThrowable, exception.getCause().getCause()); + + } + +} \ No newline at end of file diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index f3c1cd766..c9b1dce43 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -23,6 +23,7 @@ import java.net.URI; import java.net.URISyntaxException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -510,6 +511,7 @@ public class RabbitBinderTests extends endpoint = extractEndpoint(consumerBinding); container = verifyContainer(endpoint); + assertThat(container.getQueueNames()[0]).isEqualTo("foo.props.0.test"); consumerBinding.unbind(); @@ -1935,6 +1937,57 @@ public class RabbitBinderTests extends verifyAutoDeclareContextClear(binder); } + @Test + public void testErrorMessageHandlerForBatchModeDLQRepublish(TestInfo testInfo) throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(testInfo); + producerProperties.getExtension() + .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + + DirectChannel output = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + output.setBeanName("consumerBatching.Producer"); + Binding producerBinding = binder.bindProducer("c.batching.0", + output, producerProperties); + + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setBatchMode(true); + consumerProperties.getExtension().setBatchSize(2); + consumerProperties.getExtension().setEnableBatching(true); + consumerProperties.getExtension().setRepublishToDlq(true); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setDeadLetterExchange("dlqTest"); + Binding consumerBinding = binder.bindConsumer("c.batching.0", + "consumerBatching", moduleInputChannel, consumerProperties); + + List expectedMessages = List.of("message1","message2"); + + output.send(new GenericMessage<>(expectedMessages.get(0).getBytes())); + output.send(new GenericMessage<>(expectedMessages.get(1).getBytes())); + + moduleInputChannel.subscribe(message -> { + throw new RuntimeException("Exception"); + }); + + RabbitTemplate template = new RabbitTemplate( + this.rabbitTestSupport.getResource()); + long now = Instant.now().toEpochMilli(); + List deadLetterMessages = new ArrayList<>(); + while(deadLetterMessages.size() < 2 || (Instant.now().toEpochMilli() - now) > 60_000){ + deadLetterMessages.add(template + .receive("c.batching.0.consumerBatching.dlq", 10_000)); + } + + assertThat(new String(deadLetterMessages.get(0).getBody())).isIn(expectedMessages); + assertThat(new String(deadLetterMessages.get(0).getBody())).isIn(expectedMessages); + + producerBinding.unbind(); + consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); + } @SuppressWarnings("unchecked") @Test @@ -2460,7 +2513,6 @@ public class RabbitBinderTests extends verifyAutoDeclareContextClear(binder); } - @Test public void testCustomBatchingStrategy(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder();