anchor.diff:  === Converters with Schema Support
   
   The `AvroSchemaMessageConverter` supports serializing and deserializing messages either by using a predefined schema or by using the schema information available in the class (either reflectively or contained in the `SpecificRecord`).
  +If you provide a custom converter, then the default AvroSchemaMessageConverter bean is not created. The following example shows a custom converter:
  +
  +To use custom converters, you can simply add it to the application context, optionally specifying one or more `MimeTypes` with which to associate it.
  +The default `MimeType` is `application/avro`.
  +
   If the target type of the conversion is a `GenericRecord`, a schema must be set.
   
  -To use it, you can simply add it to the application context, optionally specifying one or more `MimeTypes` with which to associate it.
  -The default `MimeType` is `application/avro`.
  -
  -The following example shows how to configure a converter in a sink application by registering the Apache Avro `MessageConverter` without a predefined schema:
  +The following example shows how to configure a converter in a sink application by registering the Apache Avro `MessageConverter` without a predefined schema.
  +In this example, note that the mime type value is `avro/bytes`, not the default `application/avro`.
   
   [source,java]
   ----
  
positive.code:   return client;
  }
  ----
  NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 3.2.2.
  
  ==== Schema Registry Client Properties
  
  The Schema Registry Client supports the following properties:
  
  `spring.cloud.stream.schemaRegistryClient.endpoint`:: The location of the schema-server.
  When setting this, use a full URL, 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 in the message converter.
  Clients using the schema registry client should set this to `true`.
  +
  Default:: `true`
  
  
  === Avro Schema Registry Client Message Converters
  
  For Spring Boot applications that have a `SchemaRegistryClient` bean registered with the application context, Spring Cloud Stream autoconfigures an Apache Avro message converter that uses the schema registry client for schema management.
  This eases schema evolution, as applications that receive messages can get easy access to a writer schema that can be reconciled with their own reader schema.
  
  For outbound messages, if the content type of the channel is set to `application/*+avro`, the `MessageConverter` is activated, as shown in the following example:
  
  [source,properties]
  ----
  spring.cloud.stream.bindings.output.contentType=application/*+avro
  ----
  
  During the outbound conversion, the message converter tries to infer the schema of each outbound messages (based on its type) and register it to a subject (based on the payload type) by using the `SchemaRegistryClient`.
  If an identical schema is already found, then a reference to it is retrieved.
  If not, the schema is registered, and a new version number is provided.
  The message is sent with a `contentType` header by using the following scheme: `application/[prefix].[subject].v[version]+avro`, where `prefix` is configurable and `subject` is deduced from the payload type.
  
  For example, a message of the type `User` might be sent as a binary payload with a content type of `application/vnd.user.v2+avro`, where `user` is the subject and `2` is the version number.
  
  When receiving messages, the converter infers the schema reference from the header of the incoming message and tries to retrieve it. The schema is used as the writer schema in the deserialization process.
  
  ==== Avro Schema Registry Message Converter Properties
  
  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 by setting the following properties.
  
  spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled:: Enable if you want the converter to use reflection to infer a Schema from a POJO.
  +
  Default: `false`
  +
  spring.cloud.stream.schema.avro.readerSchema:: Avro compares schema versions by looking at a writer schema (origin payload) and a reader schema (your application payload). See the 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`
  +
  spring.cloud.stream.schema.avro.schemaLocations:: Registers any `.avsc` files listed in this property with the Schema Server.
  +
  Default: `empty`
  +
  spring.cloud.stream.schema.avro.prefix:: The prefix to be used on the Content-Type header.
  +
  Default: `vnd`
  
  
  === Schema Registration and Resolution
  
  
negative.code: To avoid any conflicts in the future, starting with 1.1.1.RELEASE, we have opted for the name `SCHEMA_REPOSITORY` for the storage table.
  Any Spring Cloud Stream 1.1.0.RELEASE users who upgrade should migrate their existing schemas to the new table before upgrading.
  
  ==== Using Confluent's Schema Registry
  
  The default configuration creates a `DefaultSchemaRegistryClient` bean.
  