Q:

Spring Cloud Stream with Avro cannot correctly convert String message

I got a problem that source sends a GenericMessage [payload=xxxxx, ...] while sink receives the message as 10,120,120,120,120,120.
This problem happened after I had set up Avro message converters. If I remove Avro message converters and use StreamListener to handle message conversion, it will work fine.
Source application.properties
spring.cloud.stream.bindings.toGreeting.destination=greeting
spring.cloud.stream.bindings.toGreeting.contentType=application/*+avro
spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true

Sink application 
server.port=8990 
spring.cloud.stream.bindings.greeting.destination=greeting

Message Converter
@Configuration
@EnableSchemaRegistryClient
public class MessageConverterConfig {
    @Bean
    public MessageConverter topic1MessageConverter() throws IOException {
        return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
    }
}

Application Class 
@SpringBootApplication
@EnableSchemaRegistryClient
public class SourceApplication {
    public static void main(String[] args) {
        SpringApplication.run(SourceApplication.class, args);
    }
}

@EnableSchemaRegistryServer
@EnableSchemaRegistryClient
@SpringBootApplication
public class SinkApplication {
    public static void main(String[] args) {
        SpringApplication.run(SinkApplication.class, args);
    }
}

Am I missing configurations?
Thanks.

A:

Here's a simple rule:
If you only want to have a message converter that can serialize/deserialize from avro, and you either provide the schema location during configuration for GenericRecords or your StreamListener method has a signature of a type of SpecificRecord. Then choose AvroSchemaMessageConverter, set it up like you did, but use avro/bytes instead. We reserve the application/*+avro for schema evolution support.
So, if you set @EnableSchemaRegistryClient then you are delegating to an external registry to have your schemas. In that case, not only you need the registry but also the schemas registered there. 
By default producers will automatically register any payload of type SpecificRecord/GenericRecord or Pojos if spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled is enabled.
In that case, the producer will actually set the header to something like application/vnd.user.v1+avro Assuming your subject is User and it's the first version.
Downstream, if your consumers are also configured with application/*+avro contentType, they will be able to read this contentType and infer it's subject/version for querying the schema server and retrieve the appropriate schema.

