From c1ce3cec7517afa716a0fb51ae1483a125e34a24 Mon Sep 17 00:00:00 2001 From: Kim Escobar Date: Tue, 22 Nov 2022 18:07:07 -0300 Subject: [PATCH] Fix ClassCastException when using batch mode Fix ClassCastException when handling error while using batch mode that prevents the republish of the message into the DLQ. The issue happened whenever the republish to dlq was enabled or if republish was disabled and max attempts were > 1 while the batch mode was enabled, because the handler was not expecting the the sourceData for the ErrorMessage could be a list of amqp message. --- ...hCapableRejectAndDontRequeueRecoverer.java | 52 ++++++++ .../rabbit/RabbitMessageChannelBinder.java | 120 +++++++++++------- ...ableRejectAndDontRequeueRecovererTest.java | 56 ++++++++ .../binder/rabbit/RabbitBinderTests.java | 54 +++++++- 4 files changed, 232 insertions(+), 50 deletions(-) create mode 100644 binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecoverer.java create mode 100644 binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/BatchCapableRejectAndDontRequeueRecovererTest.java 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();