== Spring Cloud Stream and Schema Evolution in Action with Confluent Schema Registry Server!
This repo includes thress Spring Boot applications to demonstrate Schema Evolution using Spring Cloud Stream.
Producer V1 (`producer1`), Producer V2 (`producer2`), and Consumer (`consumer`) are included in this project.
This example in particular uses the Confluent Schema Registry server.
=== 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
Make sure you are in the directory `schema-registry-confluent`
Bring up Kafka: `docker-compose up -d`
Build the project: `./mvnw clean package`
- Start the Confluent Schema Registry server in a terminal window session (Make sure that you have it locally installed)
[source,bash]
----
<ROOT_OF_CONFLUENT_PLATFORM_INSTALLATION>/bin/schema-registry-start ./etc/schema-registry/schema-registry.properties
----
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"`
If you have sensor schema already registered, you might want to consider deleting the registered schema for that to run these apps with a clean slate.
For e.g., `curl -X DELETE http://localhost:8081/subjects/sensor`
- Start `consumer` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-consumer/target/schema-registry-confluent-consumer-0.0.1-SNAPSHOT.jar
----
- Start `producer1` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-producer1/target/schema-registry-confluent-producer1-0.0.1-SNAPSHOT.jar
----
=== Sample Data
The producers are _also_ REST controllers. We will hit the `/messages` endpoint of the first producer to POST sample data.
_Example:_
[source,bash]
----
curl -X POST http://localhost:9009/messages
curl -X POST http://localhost:9009/messages
curl -X POST http://localhost:9009/messages
----
- Start `producer2` on another terminal session
[source,bash]
----
java -jar schema-registry-confluent-producer2/target/schema-registry-confluent-producer2-0.0.1-SNAPSHOT.jar
----
=== Sample Data
We will hit the `/messages` endpoint on the second producer to POST sample data.
_Example:_
[source,bash]
----
curl -X POST http://localhost:9010/messages
curl -X POST http://localhost:9010/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}
----
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, since that's the field where the `aliases` is defined in the link:https://github.com/sabbyanandan/schema/blob/master/consumer/src/main/resources/avro/sensor.avsc#L7[new schema].
=== Cleanup
Once you are done with running the samples, stop the docker containers
`docker-compose down`
and stop the Confluent Schema Registry server.
=== Using Confluent Control Center
This test can be run with the https://docs.confluent.io/current/control-center/index.html[Confluent Control Center] - a web-based tool for managing and monitoring Apache Kafka®. Control Center facilitates building and monitoring production data pipelines and streaming applications.
For a quick start you can install the control center with the help of the `docker-compose-control-center.yaml`:
[source,bash]
----
docker-compose -f ./docker-compose-control-center.yaml
----
For further info check the Confluent's https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html[Quick Start].
The https://docs.confluent.io/current/control-center/topics/schema.html[Schema Registry feature in Control Center] would help you to visualize and manage the topic schemas.
After you run the samples and post couple of messages as explained above.
1. Open the control center at `http://localhost:9021` and click on the provided cluster.
2. From the vertical menu select `Topics` tab.
3. From the list of topics select the `sensor-topic` - the topic created by the samples.
4. Click on the `Schema` tab to see the `Sensors` schema.
You can also use the Confluent Schema REST API at `http://localhost:8081`. For example the `http://localhost:8081/subjects` will list the schema names (e.g. subjects) defined.
After you have run the samples you should be able to see a schema subject name `sensor-topic-value`.
==== NOTE
Note that currently the Control Center recognizes only schema subjects created with https://docs.confluent.io/current/schema-registry/serdes-develop/index.html[TopicNameStrategy].
That means that the schema name (e.g. subject) must be named after your topic name (e.g. spring.cloud.stream.bindings.<channel>:destination) followed by `-value` suffix.
In our sample the topic (e.g. destination) is called `sensor-topic`. Therefore the schema subject should be named `sensor-topic-value` to be recognizable by the Confluent Control center.
Note that the confluent schema registry works fine with different subject naming strategies, only the control center visualization will not show them.
To configure a the schema name after the topic name yuo can use the `spring.cloud.schema.avro.subjectNamePrefix`
and `spring.cloud.schema.avro.subjectNamePrefix.subjectNamingStrategy` properties as shown below (only available with 1.1.x onwards:
[source,yaml]
----
spring:
cloud:
stream:
bindings:
process-in-0:
destination: sensor-topic
schemaRegistryClient:
endpoint: http://localhost:8081
schema:
avro:
schema-locations: classpath:avro/sensor.avsc
subjectNamePrefix: sensor-topic-value
subjectNamingStrategy: org.springframework.cloud.schema.registry.avro.SubjectPrefixOnlyNamingStrategy
server.port: 9999
----
The `subjectNamePrefix` must be set to the value of your `destination` followed by `-value` suffix. The `subjectNamingStrategy` must be set to `SubjectPrefixOnlyNamingStrategy`.
