bug: JsonToObjectTransformer throws exception when trying to determine ResolvableType

[maxtacco]

Affects Version(s): 5.2.0

Bug report

We started to get the following error after updating yo Java 13 and Spring 5.2.0.

The line the causes the error is https://github.com/spring-projects/spring-integration/blob/master/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java#L149

In this case the header value is String version of the target type and not ResolvableType instance.

This is what we use to send the message:

   @Bean
    public IntegrationFlow fileUploadFlow() {
        return IntegrationFlows.from("fileUploadChannel")
                .enrichHeaders(s -> s.headerExpressions(c ->
                        c.put("file-type", "payload.fileUpload.fileType")))
                .transform(Transformers.toJson())
                .handle(Amqp.outboundAdapter(amqpTemplate)
                        .exchangeName(FILE_UPLOAD_EXCHANGE)
                        .routingKey(FILE_UPLOAD_PROCESS_ROUTING_KEY))
                .get();
    }

And this is the code used to process the message:

    @Bean
    public IntegrationFlow fileUploadFlow(@Qualifier("fileUploadListenerContainer") final SimpleMessageListenerContainer fileUploadListenerContainer) {
        return IntegrationFlows.from(Amqp.inboundAdapter(fileUploadListenerContainer))
                .log()
                .filter("headers['x-death'] == null || headers['x-death'][0].count < 3", f -> f.discardChannel("nullChannel"))
                .transform(Transformers.fromJson(FileUploadRequest.class))
                .routeToRecipients(r ->
                        r.recipientFlow("headers['file-type'].startsWith('image/')", imageFileUploadFlow())
                                .defaultOutputChannel("nullChannel"))
                .get();
    }

2019-10-31 17:13:11.714  WARN 9674 --- [enerContainer-2] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener threw exception
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1683)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1573)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1488)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1476)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1467)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1411)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:958)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:908)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:81)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1279)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1185)
	at java.base/java.lang.Thread.run(Thread.java:830)
Caused by: org.springframework.integration.transformer.MessageTransformationException: failed to transform message; nested exception is java.lang.IllegalArgumentException: Incorrect type specified for header 'json_resolvableType'. Expected [class org.springframework.core.ResolvableType] but actual type is [class java.lang.String]
	at org.springframework.integration.transformer.AbstractTransformer.transform(AbstractTransformer.java:44)
	at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:109)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:127)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:170)
	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
	at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:444)
	at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:318)
	at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:266)
	at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:229)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:133)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:170)
	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:198)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$600(AmqpInboundChannelAdapter.java:61)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.createAndSend(AmqpInboundChannelAdapter.java:266)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:232)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1569)
	... 10 common frames omitted
Caused by: java.lang.IllegalArgumentException: Incorrect type specified for header 'json_resolvableType'. Expected [class org.springframework.core.ResolvableType] but actual type is [class java.lang.String]
	at org.springframework.messaging.MessageHeaders.get(MessageHeaders.java:215)
	at org.springframework.integration.json.JsonToObjectTransformer.obtainResolvableTypeFromHeadersIfAny(JsonToObjectTransformer.java:149)
	at org.springframework.integration.json.JsonToObjectTransformer.doTransform(JsonToObjectTransformer.java:117)
	at org.springframework.integration.transformer.AbstractTransformer.transform(AbstractTransformer.java:33)
	... 44 common frames omitted```

[artembilan]

It turns out that DefaultAmqpHeaderMapper doesn't ignore that JsonHeaders.RESOLVABLE_TYPE header when prepare an AMQP message and, therefore, stores that value as a String.

As a workaround I'd suggest to do like this:

.transform(Transformers.toJson())
.headerFilter(JsonHeaders.RESOLVABLE_TYPE)
.handle(Amqp.outboundAdapter(amqpTemplate)

Will fix tomorrow.

[maxtacco]

@artembilan That worked! Thank you for a quick solution.