You can plug KafkaAvroSerializer into KafkaProducer to send messages of Avro type to Kafka.
Currently, we support primitive types of null, Boolean, Integer,
Long, Float,
Double, String,
byte[], and complex type of IndexedRecord. Sending data of other types
to KafkaAvroSerializer will
cause a SerializationException. Typically, IndexedRecord will be used for the value of the Kafka
message. If used, the key of the Kafka message is often of one of the primitive types. When sending
a message to a topic t, the Avro schema for the key and the value will be automatically registered
in Schema Registry under the subject t-key and t-value, respectively, if the compatibility
test passes. The only exception is that the null type is never registered in Schema Registry.

In the following example, we send a message with key of type string and value of type Avro record
to Kafka. A SerializationException may occur during the send call, if the data is not well formed.

You can plug in KafkaAvroDeserializer to KafkaConsumer to receive messages of any Avro type from Kafka.
In the following example, we receive messages with key of type string and value of type Avro record
from Kafka. When getting the message key or value, a SerializationException may occur if the data is
not well formed.

The other available options that can be configured out of the box include

io.confluent.kafka.serializers.subject.RecordNameStrategy

For any Avro record type that is published to Kafka, registers the schema
in the registry under the fully-qualified record name (regardless of the
topic). This strategy allows a topic to contain a mixture of different
record types, since no intra-topic compatibility checking is performed.
Instead, checks compatibility of any occurrences of the same record name
across all topics.

io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

For any Avro record type that is published to Kafka topic <topicName>,
registers the schema in the registry under the subject name
<topicName>-<recordName>, where <recordName> is the
fully-qualified Avro record name. This strategy allows a topic to contain
a mixture of different record types, since no intra-topic compatibility
checking is performed. Moreover, different topics may contain mutually
incompatible versions of the same record name, since the compatibility
check is scoped to a particular record name within a particular topic.

You can use kafka-avro-console-producer and kafka-avro-console-consumer respectively to send and
receive Avro data in JSON format from the console. Under the hood, they use AvroMessageReader and
AvroMessageFormatter to convert between Avro and JSON.

To run the Kafka console tools, first make sure that ZooKeeper, Kafka and Schema Registry server
are all started. In the following examples, the default Schema Registry URL value is used.

You can configure that by supplying

--property schema.registry.url=address of your |sr|

in the commandline arguments of kafka-avro-console-producer and kafka-avro-console-consumer.

In the following example, we send Avro records in JSON as the message value (make sure there is no space in the schema string).

Most users can use the serializers and formatter directly and never worry about the details of how Avro messages are mapped
to bytes. However, if you're working with a language that Confluent has not developed serializers for, or simply want a deeper
understanding of how the Confluent Platform works, you may need more detail on how data is mapped to low-level bytes.

The wire format currently has only a couple of components:

Bytes

Area

Description

0

Magic Byte

Confluent serialization format version number; currently always 0.

1-4

Schema ID

4-byte schema ID as returned by Schema Registry

5-...

Data

Avro serialized data in Avro's binary encoding. The only exception is raw bytes, which
will be written directly without any special Avro encoding.

The serialization format used by Confluent Platform serializers is guaranteed to be stable over major releases without any
changes without advanced warning. This is critical because the serialization format affects how keys are mapped across
partitions. Because many applications depend on keys with the same logical format being routed to the same physical
partition, it is usually important that the physical byte format of serialized data does not change unexpectedly for an
application. Even the smallest modification can result in records with the same logical key being routed to different
partitions because messages are routed to partitions based on the hash of the key.

In order to ensure there is no variation even as the serializers are updated with new formats, the serializers are very
conservative when updating output formats. To ensure stability for clients, Confluent Platform and its serializers ensure the
following:

The format (including magic byte) will not change without significant warning over multiple Confluent Platform major
releases. Although the default may eventually be changed infrequently to allow adoption of new features by default, this
will be done very conservatively and with at least one major release between changes, during which the relevant changes
will result in user-facing warnings so no users will be caught off guard by the need for transition. Very significant,
compatibility-affecting changes will guarantee at least 1 major release of warning and 2 major releases before an
incompatible change will be made.

Within the version specified by the magic byte, the format will never change in any backwards-incompatible way. Any changes
made will be fully backward compatible with documentation in release notes and at least one version of warning will be
provided if it introduces a new serialization feature which requires additional downstream support.

Deserialization will be supported over multiple major releases. This does not guarantee indefinite support, but support for
deserializing any earlier formats will be supported indefinitely as long as there is no notified reason for
incompatibility.