1

I am using an AMQP backed channel with Spring Boot 1.4.4.RELEASE and spring-cloud-dependencies. Up to Brixton.SR6, everything is working fine. The only thing I change in my project is Brixton.SR6 to Brixton.SR7 (note that I tried all Camden versions as well and I experienced the same issue) and I start getting: 

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:873)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:783)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:703)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:98)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:189)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1236)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:684)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1190)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$500(SimpleMessageListenerContainer.java:98)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$4.doInTransaction(SimpleMessageListenerContainer.java:1157)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$4.doInTransaction(SimpleMessageListenerContainer.java:1150)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1150)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1200(SimpleMessageListenerContainer.java:98)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1363)
at java.lang.Thread.run(Thread.java:745)

Caused by: org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'Action.FieldBackup'; nested exception is org.springframework.amqp.support.converter.MessageConversionException: failed to convert to serialized Message content
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:449)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:292)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:212)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:129)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:292)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:212)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:129)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:171)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$400(AmqpInboundChannelAdapter.java:45)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$1.onMessage(AmqpInboundChannelAdapter.java:95)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:780)... 14 common frames omitted

Caused by: org.springframework.amqp.support.converter.MessageConversionException: failed to convert to serialized Message content at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:159)
at org.springframework.amqp.support.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:56)
at org.springframework.amqp.rabbit.core.RabbitTemplate.convertMessageIfNecessary(RabbitTemplate.java:1227)
at org.springframework.amqp.rabbit.core.RabbitTemplate.convertAndSend(RabbitTemplate.java:782)
at org.springframework.amqp.rabbit.core.RabbitTemplate.convertAndSend(RabbitTemplate.java:778)
at org.springframework.integration.amqp.channel.AbstractAmqpChannel.doSend(AbstractAmqpChannel.java:165)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423)
... 50 common frames omitted


Caused by: java.lang.IllegalArgumentException: Could not serialize object of type: class org.springframework.messaging.support.GenericMessage at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:52)
at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:156)
... 56 common frames omitted

Caused by: java.io.NotSerializableException: org.springframework.messaging.support.MessageHeaderAccessor
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:49)
... 57 common frames omitted

My payload is a serializable object and with debug on (in both cases, ie Brixton.SR6 and Brixton.SR7, I can see:

o.s.messaging.MessageHeaders             : Ignoring non-serializable message headers: [currentSpan, X-Current-Span]

Has anyone experienced a similar issue and found a way to upgrade to Brixton.SR7 or Camden.

The message I am trying to send to Action.FieldBackup can be seen below: 

o.s.i.amqp.channel.PollableAmqpChannel   : preSend on channel 'Action.FieldBackup', message: GenericMessage [payload=uk.gov.ons.ctp.response.action.message.instruction.ActionInstruction@2868ea3c, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=action-outbound-exchange, amqp_deliveryTag=394, X-Message-Sent=true, amqp_redelivered=true, messageSent=true, spanTraceId=15090187c236028f, spanId=8eff5c3a7c511e5b, amqp_receivedRoutingKey=Action.Field.binding, X-B3-SpanId=8eff5c3a7c511e5b, currentSpan=[Trace: 15090187c236028f, Span: 8eff5c3a7c511e5b, Parent: 15090187c236028f, exportable:false], X-B3-Sampled=0, X-B3-TraceId=15090187c236028f, id=8e4b81b8-17d0-1d15-d961-fe2b4d0e9cf5, amqp_consumerTag=amq.ctag-EXgeAZXQ5fGy8fUdHPdapQ, X-Current-Span=[Trace: 15090187c236028f, Span: 8eff5c3a7c511e5b, Parent: 15090187c236028f, exportable:false], spanSampled=0, contentType=text/xml, timestamp=1487262992385}]

And just before the exception ListenerExecutionFailedException is thrown I can see: o.s.messaging.MessageHeaders : Ignoring non-serializable message headers: [currentSpan, X-Current-Span]

My xml configuration is below: 

<bean id="actionInstructionListenerContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="queueNames" value="Action.Field" />
    <property name="concurrentConsumers" value="${messaging.consumingThreads}" />
    <property name="prefetchCount" value="${messaging.prefetchCount}" />
    <property name="txSize" value="${messaging.txSize}" />
    <property name="channelTransacted" value="true" />
    <property name="transactionManager" ref="transactionManager" />
</bean>

<!-- Xml msg is taken off the queue Action.Field and put on the channel actionInstructionXml -->
<int:channel id="actionInstructionXml" />
<bean id="simpleMessageConverter" class="org.springframework.amqp.support.converter.SimpleMessageConverter" />
<int-amqp:inbound-channel-adapter id="actionInstructionInboundAmqp" listener-container="actionInstructionListenerContainer"
                                  message-converter="simpleMessageConverter" channel="actionInstructionXml" />

<!-- Xml validated against the inbound schema -->
<int:channel id="actionInstructionInvalidXml" />
<int-amqp:outbound-channel-adapter id="invalidActionInstructionOutboundAmqp" channel="actionInstructionInvalidXml" exchange-name="action-invalid-exchange" routing-key="Action.ActionInstruction.binding" />
<!-- Option to be used in Debug/Test: throw-exception-on-rejection="true" -->
<int-xml:validating-filter id="actionInstructionValidator" input-channel="actionInstructionXml"
                           output-channel="actionInstructionXmlValid" schema-location="classpath:xsd/actionInstruction.xsd" discard-channel="actionInstructionInvalidXml"/>

<!-- The now validated xml is transformed into the jaxb generated object model -->
<!-- To mark the channel as "pollable", set the "message-driven" property to "false" -->
<int-amqp:channel id="Action.FieldBackup" message-driven="false" connection-factory="connectionFactory"
                  channel-transacted="true" transaction-manager="transactionManager" tx-size="${messaging.txSize}"
                  concurrent-consumers="${messaging.consumingThreads}" prefetch-count="${messaging.prefetchCount}" />
<int-xml:unmarshalling-transformer
        input-channel="actionInstructionXmlValid" output-channel="Action.FieldBackup"
        unmarshaller="actionInstructionMarshaller" />
4

1 回答 1

1

我们正在修复https://github.com/spring-cloud/spring-cloud-sleuth/issues/523。虽然实际上它已经在https://jira.spring.io/browse/SPR-15262中完成了。

的解决方法是：

public class ImmutableMessageInterceptor extends ChannelInterceptorAdapter {

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(message);
        return  new GenericMessage<Object>(message.getPayload(), headers.toMessageHeaders());
    }

}

您必须将其应用于您的Action.FieldBackupAMQP 支持的通道。并且按摩将从可变性钩子的剩余部分中解脱出来。

于 2017-02-21T15:20:37.997 回答