Cannot convert from [[B] to [com.example.DummyMessage] for GenericMessage

[metalpalo]

Describe the issue
After upgrade from spring-cloud-starter-parent from Hoxton.SR1 to Hoxton.SR2 we received following error when consuming kafka messages avro encoded from legacy microservices used springboot 1.5.x:

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.example.DummyMessage] for GenericMessage [payload=byte[49], headers={kafka_timestampType=CREATE_TIME, kafka_receivedTopic=Topic_DummyMessages, originalContentType=application/vnd.dummymessage.v1+avro, spanTraceId=b12de318aedd5501, spanId=b12de318aedd5501, nativeHeaders={X-B3-TraceId=[b12de318aedd5501], spanTraceId=[b12de318aedd5501], X-B3-SpanId=[b12de318aedd5501], spanId=[b12de318aedd5501], X-B3-Sampled=[0], spanSampled=[0]}, kafka_offset=11, X-B3-SpanId=b12de318aedd5501, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@398b3b8d, X-B3-Sampled=0, X-B3-TraceId=b12de318aedd5501, id=0b716343-7904-258b-d4b6-12263b6a5216, kafka_receivedPartitionId=0, spanSampled=0, contentType=application/octet-stream, kafka_receivedTimestamp=1629983108296, kafka_groupId=DummyService, timestamp=1630042306997}]
	at org.springframework.cloud.stream.config.SmartPayloadArgumentResolver.resolveArgument(SmartPayloadArgumentResolver.java:123) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
	at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:148) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:127) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:170) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:198) ~[spring-integration-core-5.2.3.RELEASE.jar:5.2.3.RELEASE]
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.sendMessageIfAny(KafkaMessageDrivenChannelAdapter.java:384) ~[spring-integration-kafka-3.2.1.RELEASE.jar:3.2.1.RELEASE]
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:75) ~[spring-integration-kafka-3.2.1.RELEASE.jar:3.2.1.RELEASE]
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:443) ~[spring-integration-kafka-3.2.1.RELEASE.jar:3.2.1.RELEASE]
	at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:417) ~[spring-integration-kafka-3.2.1.RELEASE.jar:3.2.1.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1696) ~[spring-kafka-2.3.5.RELEASE.jar:2.3.5.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1679) ~[spring-kafka-2.3.5.RELEASE.jar:2.3.5.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1634) ~[spring-kafka-2.3.5.RELEASE.jar:2.3.5.RELEASE]

The message paylods on kafka had wrapped avro contentType withing application/octet-stream:
contentType="application/octet-stream" originalContentType="application/vnd.dummymessage.v1+avro"

Both producer and consumer service has set embeddedHeaders as headerMode.

To Reproduce
Steps to reproduce the behavior:

1. Send avro kafka message from legacy SB 1.5.x microservices
2. Kafka message exists on topic
3. Start consuming SB microsevice with Hoxton.SR2
4. Consumer received error mentioned above

Version of the framework
spring-cloud-starter-parent:Hoxton.SR2
spring-boot-starter-parent: 2.2.4.RELEASE
spring-cloud-stream-binder-kafka: 3.0.2.RELEASE

Expected behavior
Kafka message should be successfully processed as for following configuration:
spring-cloud-starter-parent:Hoxton.SR1
spring-boot-starter-parent: 2.2.2.RELEASE
spring-cloud-stream-binder-kafka: 3.0.1.RELEASE

Can somebody tell me what change there? Does exist some configuration for backward compatibility?
thanks
brmetalpalo

[metalpalo]

When I hardcoded this in dependencyManagement it works and also send message outside but I afraid of imcompatible issues.

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    <version>3.0.1.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
    <version>3.0.1.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <version>3.0.1.RELEASE</version>
</dependency>

I debuged it. When using vesrion 3.0.1 the AvroSchemaRegistryClientMessageConverter has injected OriginalContentTypeResolver, but 3.0.2 has DefaultContentTypeResolver

OriginalContentTypeResolver was originally set into AvroSchemaRegistryClientMessageConverter during avro auto configuration, but later reset to DefaultContentTypeResolver in CompositeMessageConverterFactory class

The class where this happened and is different between versions and it names ContentTypeConfiguration.

1. It constructing ConfigurableCompositeMessageConverter(3.0.1) or CompositeMessageConverter(3.0.2).
2. ContentTypeConfiguration(3.0.1) sets DefaultContentTypeResolver via CompositeMessageConverterFactory only for default messages converters
3. ContentTypeConfiguration(3.0.2) sets DefaultContentTypeResolver via CompositeMessageConverterFactory also for custom converters where AvroSchemaRegistryClientMessageConverter included

Can somebody explain why this was changed there?

[metalpalo]

I have this as workaround solution working for me. But I don't know if its OK. Can somebody confirm that some problems can occur?

Slf4j
@Configuration
@ConditionalOnProperty(value = {"avro.configuration.use-original-content-type-resolver"}, matchIfMissing = true)
public class AvroConfiguration {

    @Autowired
    void fixContentTypeResolver4AvroSchemaRegistryClientMessageConverter(
            @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter converter) {
        log.info("replacing DefaultContentTypeResolver...");
        List<MessageConverter> converters = converter.getConverters();
        List<AvroSchemaRegistryClientMessageConverter> asrcmConverters = converters.stream()
                .filter(mc -> mc instanceof AvroSchemaRegistryClientMessageConverter)
                .filter(mc -> ((AvroSchemaRegistryClientMessageConverter) mc).getContentTypeResolver() instanceof DefaultContentTypeResolver)
                .map(mc -> ((AvroSchemaRegistryClientMessageConverter) mc))
                .collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(asrcmConverters)) {
            log.info("replacing DefaultContentTypeResolver with custom OriginalContentTypeResolver for AvroSchemaRegistryClientMessageConverter instances {}", asrcmConverters);
            OriginalContentTypeResolver originalContentTypeResolver = new OriginalContentTypeResolver();
            for (AvroSchemaRegistryClientMessageConverter asrcmConverter : asrcmConverters) {
                asrcmConverter.setContentTypeResolver(originalContentTypeResolver);
            }
        }
    }

    //TODO: overtaken from schema registry client because is package protected
    public static class OriginalContentTypeResolver implements ContentTypeResolver {

        private static final String BINDER_ORIGINAL_CONTENT_TYPE = "originalContentType";

        private ConcurrentMap<String, MimeType> mimeTypeCache = new ConcurrentHashMap<>();

        @Override
        public MimeType resolve(MessageHeaders headers) {
            Object contentType = headers
                    .get(BINDER_ORIGINAL_CONTENT_TYPE) != null
                    ? headers.get(BINDER_ORIGINAL_CONTENT_TYPE)
                    : headers.get(MessageHeaders.CONTENT_TYPE);
            MimeType mimeType = null;
            if (contentType instanceof MimeType) {
                mimeType = (MimeType) contentType;
            } else if (contentType instanceof String) {
                mimeType = this.mimeTypeCache.get(contentType);
                if (mimeType == null) {
                    String valueAsString = (String) contentType;
                    mimeType = MimeType.valueOf(valueAsString);
                    this.mimeTypeCache.put(valueAsString, mimeType);
                }
            }
            return mimeType;
        }
    }
}

[metalpalo]

this commit replacing resolver also for all custom converters with default one:
spring-cloud/spring-cloud-stream@cce66f3#diff-262c0de1002df243e36569ab0fcb1b8590a04fac57bc91a377221cb5e4ee7ee4

@olegz Why this was changed? If I good understood, Im not able to define custom converter with custom contentType resolver now?

[sobychacko]

@metalpalo Could you create a small reproducible sample and share that with us? Thank you!

[sobychacko]

@metalpalo Closing the issue now. Feel free to re-open with a reproducible sample if you are still facing issues with this.

[metalpalo]

Sorry for late answer, I was very busy.
I start thinking if possible to create some example, because I dont have any example avro object.

But question was simple. Why you reset contentType resolver on custom converters?
Good example is AvroSchemaRegistryClientMessageConverter, which support chunks like
contentType="application/octet-stream" originalContentType="application/vnd.dummymessage.v1+avro"
via own OriginalContentTypeResolver.

But in commit mentioned above it was reset to DefaultContentTypeResolver.

[sobychacko]

@metalpalo Could you point which commit made that change? It's not clear where that change is made, also where do you see DefaultContentTypeResolver? Seems like this is something that needs to be further debugged. If it happens with any custom converters, it may be able to be easily produced (A sample app where we can see the problem will help a lot - you may not need the AVRO converter if it happens with other custom converters). Another thing - the Spring Cloud Schema Registry project is no longer actively maintained, therefore you may not want to rely on AvroSchemaRegistryClientMessageConverter going forward. Are you using Spring Cloud Schema Registry? Kafka binder can support other schema registries such as the one from Confluent.

[metalpalo]

@sobychacko
Regarding the example, maybe original content type is used only for some binary messages, I can check later if json contentype is wrapped too, but Im not sure.

I use AvroSchemaRegistryClientMessageConverter from library, verrsion 1.0.13.RELEASE:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-schema-registry-client</artifactId>
</dependency> 

Ancestor of this class, namely AbstractAvroMessageConverter, implicitly sets OriginalContentTypeResolver as contenType resolver:

protected AbstractAvroMessageConverter(Collection<MimeType> supportedMimeTypes, AvroSchemaServiceManager manager) {
	super(supportedMimeTypes);
	setContentTypeResolver(new OriginalContentTypeResolver());
	this.avroSchemaServiceManager = manager;
}

This converter was recognized like custom one his resolver was replaced with DefaultContentTypeResolver via class CompositeMessageConverterFactory where first parameter is list of custom converters.
Before this commit, first parameter is empty ArrayList and customer converters are not modified there.
Look at following commit, class ContentTypeConfiguration :
spring-cloud/spring-cloud-stream@cce66f3#diff-262c0de1002df243e36569ab0fcb1b8590a04fac57bc91a377221cb5e4ee7ee4

[sobychacko]

I cannot locate the class DefaultContentTypeResolver in that commit or in the codebase. Also, do we need to run the schema registry from Spring for this? In order to help you further triaging this issue, I would really like to see a small sample application with steps to reproduce the issue. Could you create one on GitHub and share it with us? Or you can send a PR with the possible fix that you think is appropriate, which we can then review.

[olegz]

@metalpalo as Soby suggested several times, please provide a reproducible sample.
The originalContentType is not used by spring-cloud-stream since 2.0, so not exactly sure what is going on here. So a small reproducible sample with only a bare minimum would help us understand the issue.