"Unknown type for contentType header" after upgrading release train from Finchley to Greenwich

[justogaiht]

After changed from spring-boot-2.0.6.RELEASE and spring-cloud-Finchley.RELEASE to spring-boot-2.1.0.RELEASE and spring-cloud-Greenwich.M3,
the MessageHeader contentType changed from "application/json" to a serialized org.springframework.util.MimeType which throws an exception of NonTrustedHeaderType.

java.lang.IllegalArgumentException: Unknown type for contentType header value: class org.springframework.kafka.support.DefaultKafkaHeaderMapper$NonTrustedHeaderType
	at org.springframework.messaging.converter.DefaultContentTypeResolver.resolve(DefaultContentTypeResolver.java:76)
	at org.springframework.messaging.converter.AbstractMessageConverter.getMimeType(AbstractMessageConverter.java:249)
	at org.springframework.messaging.converter.AbstractMessageConverter.supportsMimeType(AbstractMessageConverter.java:235)
	at org.springframework.messaging.converter.MappingJackson2MessageConverter.canConvertFrom(MappingJackson2MessageConverter.java:146)
	at org.springframework.messaging.converter.AbstractMessageConverter.fromMessage(AbstractMessageConverter.java:178)
	at org.springframework.messaging.converter.CompositeMessageConverter.fromMessage(CompositeMessageConverter.java:70)
	at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:137)
	at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:114)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:145)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:114)
	at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:162)
	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:451)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:400)
	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:203)
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.sendMessageIfAny(KafkaMessageDrivenChannelAdapter.java:369)
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$400(KafkaMessageDrivenChannelAdapter.java:74)
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:431)
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:402)
	at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.lambda$onMessage$0(RetryingMessageListenerAdapter.java:120)
	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:211)
	at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:114)
	at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:40)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1141)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1115)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1057)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:903)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:753)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
	at java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.lang.Thread.run(Thread.java:745)

After add the package "org.springframework.util" to the trusted packages of DefaultKafkaHeaderMapper :

ERROR o.s.k.s.DefaultKafkaHeaderMapper - Could not decode json type: {"type":"application","subtype":"json","parameters":{},"concrete":true,"wildcardType":false,"wildcardSubtype":false,"charset":null} for key: contentType
com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of `org.springframework.util.MimeType` (although at least one Creator exists): cannot deserialize from Object value (no delegate- or property-based Creator)
 at [Source: (byte[])"{"type":"application","subtype":"json","parameters":{},"concrete":true,"wildcardType":false,"wildcardSubtype":false,"charset":null}"; line: 1, column: 2]

[sobychacko]

@justogaiht Can you specify more details on how we can reproduce the issue? Please provide instructions or provide sample code and configuration so that we can further troubleshoot.

[garyrussell]

@sobychacko I think it's caused by this.

So, if you have an "older" producer that doesn't send it as a String, the "new" consumer (with the back-ported header mapper) will try to decode it as a MimeType.

cc/ @artembilan - we should reinstate this in spring-kafka 2.2 to enable backward compatibility (a 2.1 producer and 2.2 consumer).

@sobychacko I think you will have the same problem in 2.0 because we backported the 2.2 header mapper, we ended up removing the custom MimeTypeJsonDeserializer.

I think all you need to do is restore those two lines in the back-ported header mapper to fix it in 2.0. You could also fix it in 2.1 by using your version of the header mapper there (and adding the custom deserializer).

[artembilan]

M-m-m. Is there be enough to add org.springframework.util and org.springframework.http to trusted packages?

[garyrussell]

No; there is no default CTOR - it's not Jackson-friendly

[artembilan]

OK. Then MimeTypeJsonDeserializer back!