spring-cloud-stream-samples/schema-registry-samples/schema-registry-confluent-avro-serializer

== Spring Cloud Stream and Schema Evolution in Action with Confluent Schema Registry Server and Confluent Avro Serializers!

These are collection of Spring Boot applications to demonstrate Schema Evolution using Spring Cloud Stream and Confluent Schema Registry.
Producer V1 (`producer1`), Producer V2 (`producer2`), and Consumer (`consumer`) are included in this project.

These samples are identical in functionality with https://github.com/spring-cloud/spring-cloud-stream-samples/tree/master/schema-registry-samples/schema-registry-confluent[this sample],
but with an important difference. All the components (producers and consumer) use the Avro serializer provided by Confluent, rather than using the Spring Cloud Stream provided Avro serializers.
Spring Cloud Stream producers in this sample are using native encoding to use the Confluent avro serializer and similarly the consumer is using native decoding to use the Confluent Avro deserializer.
The benefit is that these components are now cross compatible with external tools that use the Confluent Avro serializers such
as the out of the box tools - `kafka-avro-console-consumer` and `kafka-avro-console-producer` - that come with Confluent Schema registry.
If your use case requires to use these external tools, then we recommend using the strategies implemented in these samples.

=== Requirement
As a developer, I'd like to design my consumer to be resilient to differing payload schemas and I want to use Confluent Schema Registry.

=== Assumptions
There are a lot of online literature on Schema Evolution, so we are going to skip defining them here. For this demonstration,
however, we will simply assume there are two producers producing events with different payload schemas. A consumer that
consumes both the payload versions will be designed to adapt to evolving schemas.

=== Running the application

Start the Confluent Platform:

```
git clone https://github.com/confluentinc/cp-docker-images
cd cp-docker-images
git checkout 5.2.1-post
cd examples/cp-all-in-one/
docker-compose up -d --build
```

Make sure you are in the directory `schema-registry-confluent-avro-serializer`

In order to run this sample, you need to set compatibility to `NONE` on Confluent schema registry server.

`curl -X PUT http://127.0.0.1:8081/config -d '{"compatibility": "NONE"}' -H "Content-Type:application/json"`

- Start `consumer` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-avro-serializer-consumer/target/schema-registry-confluent-avro-serializer-consumer-0.0.1-SNAPSHOT.jar
----
- Start `producer1` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-avro-serializer-producer1/target/schema-registry-confluent-avro-serializer-producer1-0.0.1-SNAPSHOT.jar
----
- Start `producer2` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-avro-serializer-producer2/target/schema-registry-confluent-avro-serializer-producer2-0.0.1-SNAPSHOT.jar
----

=== Sample Data
Both the producers in the demonstration are _also_ REST controllers. We will hit the `/messages` endpoint on each producer
to POST sample data.

_Example:_
[source,bash]
----
curl -X POST http://localhost:9009/messages
curl -X POST http://localhost:9010/messages
curl -X POST http://localhost:9009/messages
curl -X POST http://localhost:9009/messages
curl -X POST http://localhost:9010/messages
----

=== Output
The consumer should log the results.

[source,bash,options=nowrap,subs=attributes]
----
{"id": "d135efc3-72f8-4612-9497-184cae508e31-v1", "internalTemperature": 34.36362, "externalTemperature": 0.0, "acceleration": 9.656547, "velocity": 33.29733}
{"id": "fd2467ce-ae09-4fd4-9cde-d9ff33fac89b-v2", "internalTemperature": 34.840473, "externalTemperature": 0.0, "acceleration": 9.709609, "velocity": 23.046476}
{"id": "4ac70c32-9ffe-4c90-914a-fa28024f5faa-v1", "internalTemperature": 23.74807, "externalTemperature": 0.0, "acceleration": 7.5003176, "velocity": 15.848035}
{"id": "3ecaae18-3144-4570-800a-223ca3198001-v1", "internalTemperature": 28.410656, "externalTemperature": 0.0, "acceleration": 1.752817, "velocity": 69.82016}
{"id": "149637a9-c7a6-4ab8-b7aa-021c72d9ebd7-v2", "internalTemperature": 2.2332578, "externalTemperature": 0.0, "acceleration": 6.251889, "velocity": 65.84996}
----

You can also sue external tools to query:

----
./bin/kafka-avro-console-consumer --topic sensor-topic \
--bootstrap-server localhost:9092 \
--from-beginning
{"id":"a4c03b84-3598-4b29-8507-edf05f211263-v1","temperature":7.0681753,"acceleration":6.8061967,"velocity":86.663795}
{"id":"92e5fa92-c90b-49ca-b3c1-1b0b98fb3d82-v1","temperature":3.0760436,"acceleration":4.700919,"velocity":20.379478}
{"id":"0ac79b9f-6ba3-4381-b933-da8355555650-v1","temperature":21.31792,"acceleration":7.2651076,"velocity":14.394546}
{"id":"c6c6a453-b8a0-4c67-ab5d-be4f6f04123a-v2","internalTemperature":31.67511,"externalTemperature":0.0,"acceleration":3.66884,"velocity":80.335815,"accelerometer":null,"magneticField":null}
----

NOTE: Refer to the payload suffix in the `id` field. Each of them are appended with `-v1` or `-v2` indicating they are from
`producer1` and `producer2` respectively.

=== What just happened?
The schema evolved on the `temperature` field. That field is now split into `internalTemperature` and `externalTemperature`,
as two separate fields. The `producer1` produces payload only with `temperature` and on the other hand, `producer2` produces
payload with `internalTemperature` and `externalTemperature` fields in it.

The `consumer` is coded against a base schema that include the split fields.

The `consumer` app can happily deserialize the payload with `internalTemperature` and `externalTemperature` fields. However, when
a `producer1` payload arrives (which includes `temperature` field), the schema evolution and compatibility check are automatically
applied.

Because each payload also includes the payload version in the header, Spring Cloud Stream with the help of Schema
Registry server and Avro, the schema evolution occurs behind the scenes. The automatic mapping of `temperature` to
`internalTemperature` field is applied.

=== Cleanup

Once you are done with running the samples, stop the docker containers

`docker-compose down`