anchor.diff:  
   === Apache Avro Message Converters
   
  -The `spring-cloud-stream-schema` module contains two  types of message converters that can be used for Apache Avro serialization:
  +The `spring-cloud-stream-schema` module contains two types of message converters that can be used for Apache Avro serialization:
   
   *     converters using the class information of the serialized/deserialized objects, or a schema with a location known at startup;
   *     converters using a schema registry - they locate the schemas at runtime, as well as dynamically registering new schemas as domain objects evolve.
  
positive.code: 
  If you have enabled Avro based schema registry client by setting `spring.cloud.stream.bindings.output.contentType=application/*+avro` you can customize the behavior of the registration with the following properties.
  
  dynamicSchemaGenerationEnabled:: Enable if you want the converter to use reflection to infer a Schema from a POJO.
  +
  Default:: `false`
  +
  readerSchema:: Avro compares schema versions by looking at a writer schema (origin payload) and a reader schema (your application payload), check https://avro.apache.org/docs/1.7.6/spec.html[Avro] documentation for more information. If set, this overrides any lookups at the schema server and uses the local schema as the reader schema.
  Default:: `null`
  +
  schemaLocations:: Register any `.avsc` files listed in this property with the Schema Server.
  +
  Default:: `empty`
  +
  prefix:: The prefix to be used on the Content-Type header
  +
  Default:: `vnd`
  +
  
  === How Schema Registration works in Spring Cloud Stream
  
  To better understand how does Spring Cloud Stream registers new schemas, and uses Avro schema comparison we split the process into two steps described below, one for the registration part, and one for the resolution of schemas.
  
  
  ==== Schema Registration Process (Serialization)
  
  The first part of the registration process is extracting a schema from the payload that is being sent over a channel.
  If you are using Avro types such as SpecificRecord or GenericRecord, those types already contain the schema so we just access that, in case of POJOs we only infer a schema if the property `spring.cloud.stream.schema.avro/dynamicSchemaGenerationEnabled` is set to `true`.
  
  
  .Schema Writer Resolution Process
  image::schema_resolution.png[width=300,scaledwidth="75%",align="center"]
  
  
  Once a schema is obtained, the converter will then load it's metadata (version) from the remote server.
  First it queries a local cache, and if not found it then submits the data to the server that will reply with versioning information.
  We always cache the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.
  
  .Schema Registration Process
  image::registration.png[width=300,scaledwidth="75%",align="center"]
  
  With the schema version information, the converter sets the content type of the message to carry the version information such as `application/vnd.user.v1+avro`
  
  ==== Schema Resolution Process (Deserialization)
  
  When reading messages that contains version information, the converter will query the Schema server to fetch the *writer* schema of the message.
  Once it has found the correct schema of the incoming message, it then resolves the reader schema and using Avro evolution support reads it into the reader definition (setting defaults and missing properties)
  
  .Schema Reading Resolution Process
  image::schema_reading.png[width=300,scaledwidth="75%",align="center"]
  
negative.code: [NOTE]
  ====
  The default converter is optimized to cache not only the schemas from the remote server but also the `parse()` and `toString()` methods that are quite expensive.
  Because of this, it uses a `DefaultSchemaRegistryClient` that does not caches responses.
  If you intend to use the client directly on your code, you can request a bean that also caches responses to be created.
  To do that, just add the property `spring.cloud.stream.schemaRegistryClient.cached=true` to your application properties.
  ====
  
  ==== Schema Registry Client properties
  
  The Schema Registry Client supports the following properties:
  
  spring.cloud.stream.schemaRegistryClient.endpoint:: The location of the schema-server.
  Use a full URL when setting this, including protocol (`http` or `https`) , port and context path.
  +
  Default:: ``http://localhost:8990/``
  spring.cloud.stream.schemaRegistryClient.cached:: Whether the client should cache schema server responses.
  Normally set to `false`, as the caching happens iin the message converter.
  Clients using the schema registry client should set this to `true`.
  +
  Default:: `true`
  
  
  === Avro Schema Registry Client Message Converters
  