SpringCloud/spring-cloud-schema-registry
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Initial commit

Browse Source 
Migrating schema registry code from Spring Cloud Stream core into its own code base
This commit is contained in:
Soby Chacko
2019-08-16 19:11:41 -04:00
commit fb3eabc669
89 changed files with 7070 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
docs/src/main/asciidoc/README.adoc Normal file
View File

@@ -0,0 +1,42 @@
== Spring Cloud Schema Registry
When organizations have a messaging based pub/sub architecture and multiple producer and consumer microservices communicate each other, it is often necessary for all those microservices to agree on a contract that is based on a schema.
When such a schema needs to evolve to accommodate new business requirements, the existing components are still required to continue to work.
This project provides support for a standalone schema registry server using which aforementioned schema can be registered and used by the applications.
It also contains support for avro based schema registry clients, which essentially provide message converters that communicates with the schema registry for reconciling schema during message conversion.
The schema evolution support provided by this project works both with the aforementioned standalone schema registry as well as the scheam registry provided by Confluent that specifically works with Apache Kafka.
==== Spring Cloud Schema Registry overview
Spring Cloud Schema Registry provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. In certain cases, the schema can be inferred from the payload type on serialization or from the target type on deserialization.
However, many applications benefit from having access to an explicit schema that describes the binary data format.
A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format.
A schema is referenceable as a tuple consisting of:
* A subject that is the logical name of the schema
* The schema version
* The schema format, which describes the binary format of the data
Spring Cloud Schema Registry provides the following compoents
* Standalone Schema Registry Server
By default, it is using an H2 database, but server can be used with other databases by providing appropriate datasource configuration.
* Schema registry clients capable of message marshalling by communicating with a Schema Registry.
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.
== Project page
You can read more about Spring Cloud Schema Registry by going to https://spring.io/projects/spring-cloud-schema-registry[the project page]
== Building
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/building.adoc[]
== Contributing
include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/contributing.adoc[]

330
docs/src/main/asciidoc/ghpages.sh Executable file
View File

@@ -0,0 +1,330 @@
#!/bin/bash -x
set -e
# Set default props like MAVEN_PATH, ROOT_FOLDER etc.
function set_default_props() {
# The script should be executed from the root folder
ROOT_FOLDER=`pwd`
echo "Current folder is ${ROOT_FOLDER}"
if [[ ! -e "${ROOT_FOLDER}/.git" ]]; then
echo "You're not in the root folder of the project!"
exit 1
fi
# Prop that will let commit the changes
COMMIT_CHANGES="no"
MAVEN_PATH=${MAVEN_PATH:-}
echo "Path to Maven is [${MAVEN_PATH}]"
REPO_NAME=${PWD##*/}
echo "Repo name is [${REPO_NAME}]"
SPRING_CLOUD_STATIC_REPO=${SPRING_CLOUD_STATIC_REPO:-git@github.com:spring-cloud/spring-cloud-static.git}
echo "Spring Cloud Static repo is [${SPRING_CLOUD_STATIC_REPO}"
}
# Check if gh-pages exists and docs have been built
function check_if_anything_to_sync() {
git remote set-url --push origin `git config remote.origin.url | sed -e 's/^git:/https:/'`
if ! (git remote set-branches --add origin gh-pages && git fetch -q); then
echo "No gh-pages, so not syncing"
exit 0
fi
if ! [ -d docs/target/generated-docs ] && ! [ "${BUILD}" == "yes" ]; then
echo "No gh-pages sources in docs/target/generated-docs, so not syncing"
exit 0
fi
}
function retrieve_current_branch() {
# Code getting the name of the current branch. For master we want to publish as we did until now
# https://stackoverflow.com/questions/1593051/how-to-programmatically-determine-the-current-checked-out-git-branch
# If there is a branch already passed will reuse it - otherwise will try to find it
CURRENT_BRANCH=${BRANCH}
if [[ -z "${CURRENT_BRANCH}" ]] ; then
CURRENT_BRANCH=$(git symbolic-ref -q HEAD)
CURRENT_BRANCH=${CURRENT_BRANCH##refs/heads/}
CURRENT_BRANCH=${CURRENT_BRANCH:-HEAD}
fi
echo "Current branch is [${CURRENT_BRANCH}]"
git checkout ${CURRENT_BRANCH} || echo "Failed to check the branch... continuing with the script"
}
# Switches to the provided value of the release version. We always prefix it with `v`
function switch_to_tag() {
git checkout v${VERSION}
}
# Build the docs if switch is on
function build_docs_if_applicable() {
if [[ "${BUILD}" == "yes" ]] ; then
./mvnw clean install -P docs -pl docs -DskipTests
fi
}
# Get the name of the `docs.main` property
# Get whitelisted branches - assumes that a `docs` module is available under `docs` profile
function retrieve_doc_properties() {
MAIN_ADOC_VALUE=$("${MAVEN_PATH}"mvn -q \
-Dexec.executable="echo" \
-Dexec.args='${docs.main}' \
--non-recursive \
org.codehaus.mojo:exec-maven-plugin:1.3.1:exec)
echo "Extracted 'main.adoc' from Maven build [${MAIN_ADOC_VALUE}]"
WHITELIST_PROPERTY=${WHITELIST_PROPERTY:-"docs.whitelisted.branches"}
WHITELISTED_BRANCHES_VALUE=$("${MAVEN_PATH}"mvn -q \
-Dexec.executable="echo" \
-Dexec.args="\${${WHITELIST_PROPERTY}}" \
org.codehaus.mojo:exec-maven-plugin:1.3.1:exec \
-P docs \
-pl docs)
echo "Extracted '${WHITELIST_PROPERTY}' from Maven build [${WHITELISTED_BRANCHES_VALUE}]"
}
# Stash any outstanding changes
function stash_changes() {
git diff-index --quiet HEAD && dirty=$? || (echo "Failed to check if the current repo is dirty. Assuming that it is." && dirty="1")
if [ "$dirty" != "0" ]; then git stash; fi
}
# Switch to gh-pages branch to sync it with current branch
function add_docs_from_target() {
local DESTINATION_REPO_FOLDER
if [[ -z "${DESTINATION}" && -z "${CLONE}" ]] ; then
DESTINATION_REPO_FOLDER=${ROOT_FOLDER}
elif [[ "${CLONE}" == "yes" ]]; then
mkdir -p ${ROOT_FOLDER}/target
local clonedStatic=${ROOT_FOLDER}/target/spring-cloud-static
if [[ ! -e "${clonedStatic}/.git" ]]; then
echo "Cloning Spring Cloud Static to target"
git clone ${SPRING_CLOUD_STATIC_REPO} ${clonedStatic} && git checkout gh-pages
else
echo "Spring Cloud Static already cloned - will pull changes"
cd ${clonedStatic} && git checkout gh-pages && git pull origin gh-pages
fi
DESTINATION_REPO_FOLDER=${clonedStatic}/${REPO_NAME}
mkdir -p ${DESTINATION_REPO_FOLDER}
else
if [[ ! -e "${DESTINATION}/.git" ]]; then
echo "[${DESTINATION}] is not a git repository"
exit 1
fi
DESTINATION_REPO_FOLDER=${DESTINATION}/${REPO_NAME}
mkdir -p ${DESTINATION_REPO_FOLDER}
echo "Destination was provided [${DESTINATION}]"
fi
cd ${DESTINATION_REPO_FOLDER}
git checkout gh-pages
git pull origin gh-pages
# Add git branches
###################################################################
if [[ -z "${VERSION}" ]] ; then
copy_docs_for_current_version
else
copy_docs_for_provided_version
fi
commit_changes_if_applicable
}
# Copies the docs by using the retrieved properties from Maven build
function copy_docs_for_current_version() {
if [[ "${CURRENT_BRANCH}" == "master" ]] ; then
echo -e "Current branch is master - will copy the current docs only to the root folder"
for f in docs/target/generated-docs/*; do
file=${f#docs/target/generated-docs/*}
if ! git ls-files -i -o --exclude-standard --directory | grep -q ^$file$; then
# Not ignored...
cp -rf $f ${ROOT_FOLDER}/
git add -A ${ROOT_FOLDER}/$file
fi
done
COMMIT_CHANGES="yes"
else
echo -e "Current branch is [${CURRENT_BRANCH}]"
# https://stackoverflow.com/questions/29300806/a-bash-script-to-check-if-a-string-is-present-in-a-comma-separated-list-of-strin
if [[ ",${WHITELISTED_BRANCHES_VALUE}," = *",${CURRENT_BRANCH},"* ]] ; then
mkdir -p ${ROOT_FOLDER}/${CURRENT_BRANCH}
echo -e "Branch [${CURRENT_BRANCH}] is whitelisted! Will copy the current docs to the [${CURRENT_BRANCH}] folder"
for f in docs/target/generated-docs/*; do
file=${f#docs/target/generated-docs/*}
if ! git ls-files -i -o --exclude-standard --directory | grep -q ^$file$; then
# Not ignored...
# We want users to access 2.0.0.BUILD-SNAPSHOT/ instead of 1.0.0.RELEASE/spring-cloud.sleuth.html
if [[ "${file}" == "${MAIN_ADOC_VALUE}.html" ]] ; then
# We don't want to copy the spring-cloud-sleuth.html
# we want it to be converted to index.html
cp -rf $f ${ROOT_FOLDER}/${CURRENT_BRANCH}/index.html
git add -A ${ROOT_FOLDER}/${CURRENT_BRANCH}/index.html
else
cp -rf $f ${ROOT_FOLDER}/${CURRENT_BRANCH}
git add -A ${ROOT_FOLDER}/${CURRENT_BRANCH}/$file
fi
fi
done
COMMIT_CHANGES="yes"
else
echo -e "Branch [${CURRENT_BRANCH}] is not on the white list! Check out the Maven [${WHITELIST_PROPERTY}] property in
[docs] module available under [docs] profile. Won't commit any changes to gh-pages for this branch."
fi
fi
}
# Copies the docs by using the explicitly provided version
function copy_docs_for_provided_version() {
local FOLDER=${DESTINATION_REPO_FOLDER}/${VERSION}
mkdir -p ${FOLDER}
echo -e "Current tag is [v${VERSION}] Will copy the current docs to the [${FOLDER}] folder"
for f in ${ROOT_FOLDER}/docs/target/generated-docs/*; do
file=${f#${ROOT_FOLDER}/docs/target/generated-docs/*}
copy_docs_for_branch ${file} ${FOLDER}
done
COMMIT_CHANGES="yes"
CURRENT_BRANCH="v${VERSION}"
}
# Copies the docs from target to the provided destination
# Params:
# $1 - file from target
# $2 - destination to which copy the files
function copy_docs_for_branch() {
local file=$1
local destination=$2
if ! git ls-files -i -o --exclude-standard --directory | grep -q ^${file}$; then
# Not ignored...
# We want users to access 2.0.0.BUILD-SNAPSHOT/ instead of 1.0.0.RELEASE/spring-cloud.sleuth.html
if [[ ("${file}" == "${MAIN_ADOC_VALUE}.html") || ("${file}" == "${REPO_NAME}.html") ]] ; then
# We don't want to copy the spring-cloud-sleuth.html
# we want it to be converted to index.html
cp -rf $f ${destination}/index.html
git add -A ${destination}/index.html
else
cp -rf $f ${destination}
git add -A ${destination}/$file
fi
fi
}
function commit_changes_if_applicable() {
if [[ "${COMMIT_CHANGES}" == "yes" ]] ; then
COMMIT_SUCCESSFUL="no"
git commit -a -m "Sync docs from ${CURRENT_BRANCH} to gh-pages" && COMMIT_SUCCESSFUL="yes" || echo "Failed to commit changes"
# Uncomment the following push if you want to auto push to
# the gh-pages branch whenever you commit to master locally.
# This is a little extreme. Use with care!
###################################################################
if [[ "${COMMIT_SUCCESSFUL}" == "yes" ]] ; then
git push origin gh-pages
fi
fi
}
# Switch back to the previous branch and exit block
function checkout_previous_branch() {
# If -version was provided we need to come back to root project
cd ${ROOT_FOLDER}
git checkout ${CURRENT_BRANCH} || echo "Failed to check the branch... continuing with the script"
if [ "$dirty" != "0" ]; then git stash pop; fi
exit 0
}
# Assert if properties have been properly passed
function assert_properties() {
echo "VERSION [${VERSION}], DESTINATION [${DESTINATION}], CLONE [${CLONE}]"
if [[ "${VERSION}" != "" && (-z "${DESTINATION}" && -z "${CLONE}") ]] ; then echo "Version was set but destination / clone was not!"; exit 1;fi
if [[ ("${DESTINATION}" != "" && "${CLONE}" != "") && -z "${VERSION}" ]] ; then echo "Destination / clone was set but version was not!"; exit 1;fi
if [[ "${DESTINATION}" != "" && "${CLONE}" == "yes" ]] ; then echo "Destination and clone was set. Pick one!"; exit 1;fi
}
# Prints the usage
function print_usage() {
cat <<EOF
The idea of this script is to update gh-pages branch with the generated docs. Without any options
the script will work in the following manner:
- if there's no gh-pages / target for docs module then the script ends
- for master branch the generated docs are copied to the root of gh-pages branch
- for any other branch (if that branch is whitelisted) a subfolder with branch name is created
and docs are copied there
- if the version switch is passed (-v) then a tag with (v) prefix will be retrieved and a folder
with that version number will be created in the gh-pages branch. WARNING! No whitelist verification will take place
- if the destination switch is passed (-d) then the script will check if the provided dir is a git repo and then will
switch to gh-pages of that repo and copy the generated docs to `docs/<project-name>/<version>`
- if the destination switch is passed (-d) then the script will check if the provided dir is a git repo and then will
switch to gh-pages of that repo and copy the generated docs to `docs/<project-name>/<version>`
USAGE:
You can use the following options:
-v|--version - the script will apply the whole procedure for a particular library version
-d|--destination - the root of destination folder where the docs should be copied. You have to use the full path.
E.g. point to spring-cloud-static folder. Can't be used with (-c)
-b|--build - will run the standard build process after checking out the branch
-c|--clone - will automatically clone the spring-cloud-static repo instead of providing the destination.
Obviously can't be used with (-d)
EOF
}
# ==========================================
# ____ ____ _____ _____ _____ _______
# / ____|/ ____| __ \|_ _| __ \__ __|
# | (___ | | | |__) | | | | |__) | | |
# \___ \| | | _ / | | | ___/ | |
# ____) | |____| | \ \ _| |_| | | |
# |_____/ \_____|_| \_\_____|_| |_|
#
# ==========================================
while [[ $# > 0 ]]
do
key="$1"
case ${key} in
-v|--version)
VERSION="$2"
shift # past argument
;;
-d|--destination)
DESTINATION="$2"
shift # past argument
;;
-b|--build)
BUILD="yes"
;;
-c|--clone)
CLONE="yes"
;;
-h|--help)
print_usage
exit 0
;;
*)
echo "Invalid option: [$1]"
print_usage
exit 1
;;
esac
shift # past argument or value
done
assert_properties
set_default_props
check_if_anything_to_sync
if [[ -z "${VERSION}" ]] ; then
retrieve_current_branch
else
switch_to_tag
fi
build_docs_if_applicable
retrieve_doc_properties
stash_changes
add_docs_from_target
checkout_previous_branch

BIN
docs/src/main/asciidoc/images/schema_reading.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 45 KiB

BIN
docs/src/main/asciidoc/images/schema_resolution.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 27 KiB

16
docs/src/main/asciidoc/index.adoc Normal file
View File

@@ -0,0 +1,16 @@
= Spring Cloud Schema Registry Reference Documentation
Unascribed
*{spring-cloud-schema-registry-version}*
:docinfo: shared
The reference documentation consists of the following sections:
[horizontal]
<<spring-cloud-schema-registry.adoc#,Reference Guide>> :: Spring Cloud Schema Registry Reference
Relevant Links:
[horizontal]
https://spring.io/projects/spring-cloud-stream[Spring Cloud Stream] :: Spring Cloud Stream

25
docs/src/main/asciidoc/sagan-index.adoc Normal file
View File

@@ -0,0 +1,25 @@
Spring Cloud Schema Registry overview
Spring Cloud Schema Registry provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. In certain cases, the schema can be inferred from the payload type on serialization or from the target type on deserialization.
However, many applications benefit from having access to an explicit schema that describes the binary data format.
A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format.
A schema is referenceable as a tuple consisting of:
* A subject that is the logical name of the schema
* The schema version
* The schema format, which describes the binary format of the data
Spring Cloud Schema Registry provides the following compoents
* Standalone Schema Registry Server
By default, it is using an H2 database, but server can be used with other databases by providing appropriate datasource configuration.
* Schema registry clients capable of message marshalling by communicating with a Schema Registry.
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.

366
docs/src/main/asciidoc/spring-cloud-schema-registry.adoc Normal file
View File

@@ -0,0 +1,366 @@
= Spring Cloud Schema Registry
Unascribed
*{spring-cloud-schema-registry-version}*
---
:github: https://github.com/spring-cloud/spring-cloud-schema-registry
:githubmaster: {github}/tree/master
:docslink: {githubmaster}/docs/src/main/asciidoc
:nofooter:
== Introduction
When organizations have a messaging based pub/sub architecture and multiple producer and consumer microservices communicate each other, it is often necessary for all those microservices to agree on a contract that is based on a schema.
When such a schema needs to evolve to accommodate new business requirements, the existing components are still required to continue to work.
This project provides support for a standalone schema registry server using which aforementioned schema can be registered and used by the applications.
It also contains support for avro based schema registry clients, which essentially provide message converters that communicates with the schema registry for reconciling schema during message conversion.
The schema evolution support provided by this project works both with the aforementioned standalone schema registry as well as the scheam registry provided by Confluent that specifically works with Apache Kafka.
==== Spring Cloud Schema Registry overview
Spring Cloud Schema Registry provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. In certain cases, the schema can be inferred from the payload type on serialization or from the target type on deserialization.
However, many applications benefit from having access to an explicit schema that describes the binary data format.
A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format.
A schema is referenceable as a tuple consisting of:
* A subject that is the logical name of the schema
* The schema version
* The schema format, which describes the binary format of the data
Spring Cloud Schema Registry provides the following compoents
* Standalone Schema Registry Server
By default, it is using an H2 database, but server can be used with other databases by providing appropriate datasource configuration.
* Schema registry clients capable of message marshalling by communicating with a Schema Registry.
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.
=== Schema Registry Client
The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, which has the following structure:
[source,java]
----
public interface SchemaRegistryClient {
SchemaRegistrationResponse register(String subject, String format, String schema);
String fetch(SchemaReference schemaReference);
String fetch(Integer id);
}
----
Spring Cloud Stream provides out-of-the-box implementations for interacting with its own schema server and for interacting with the Confluent Schema Registry.
A client for the Spring Cloud Stream schema registry can be configured by using the `@EnableSchemaRegistryClient`, as follows:
[source,java]
----
@EnableBinding(Sink.class)
@SpringBootApplication
@EnableSchemaRegistryClient
public static class AvroSinkApplication {
...
}
----
NOTE: The default converter is optimized to cache not only the schemas from the remote server but also the `parse()` and `toString()` methods, which are quite expensive.
Because of this, it uses a `DefaultSchemaRegistryClient` that does not cache responses.
If you intend to change the default behavior, you can use the client directly on your code and override it to the desired outcome.
To do so, you have to 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.
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:: `false`
=== Avro Schema Registry Client Message Converters
For applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream auto configures an Apache Avro message converter 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`
spring.cloud.stream.schema.avro.subjectNamingStrategy:: Determines the subject name used to register the Avro schema in the schema registry. Two implementations are available, `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy`,
where the subject is the schema name, and `org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy`, which returns a fully qualified subject using the Avro schema namespace and name. Custom strategies can be created by implementing `org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy`.
+
Default: `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy`
=== Apache Avro Message Converters
Spring Cloud Stream provides support for schema-based message converters through its `spring-cloud-stream-schema` module.
Currently, the only serialization format supported out of the box for schema-based message converters is Apache Avro, with more formats to be added in future versions.
The `spring-cloud-stream-schema` module contains two types of message converters that can be used for Apache Avro serialization:
* Converters that use the class information of the serialized or deserialized objects or a schema with a location known at startup.
* Converters that use a schema registry. They locate the schemas at runtime and dynamically register new schemas as domain objects evolve.
=== 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.
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]
----
@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {
...
@Bean
public MessageConverter userMessageConverter() {
return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
}
}
----
Conversely, the following application registers a converter with a predefined schema (found on the classpath):
[source,java]
----
@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {
...
@Bean
public MessageConverter userMessageConverter() {
AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
converter.setSchemaLocation(new ClassPathResource("schemas/User.avro"));
return converter;
}
}
----
=== Schema Registry Server
Spring Cloud Stream provides a schema registry server implementation.
To use it, you can add the `spring-cloud-stream-schema-server` artifact to your project and use the `@EnableSchemaRegistryServer` annotation, which adds the schema registry server REST controller to your application.
This annotation is intended to be used with Spring Boot web applications, and the listening port of the server is controlled by the `server.port` property.
The `spring.cloud.stream.schema.server.path` property can be used to control the root path of the schema server (especially when it is embedded in other applications).
The `spring.cloud.stream.schema.server.allowSchemaDeletion` boolean property enables the deletion of a schema. By default, this is disabled.
The schema registry server uses a relational database to store the schemas.
By default, it uses an embedded database.
You can customize the schema storage by using the http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-sql[Spring Boot SQL database and JDBC configuration options].
The following example shows a Spring Boot application that enables the schema registry:
[source,java]
----
@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaRegistryServerApplication {
public static void main(String[] args) {
SpringApplication.run(SchemaRegistryServerApplication.class, args);
}
}
----
==== Schema Registry Server API
The Schema Registry Server API consists of the following operations:
* `POST /` -- see `<<spring-cloud-stream-overview-registering-new-schema>>`
* 'GET /{subject}/{format}/{version}' -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format-version>>`
* `GET /{subject}/{format}` -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format>>`
* `GET /schemas/{id}` -- see `<<spring-cloud-stream-overview-retrieve-schema-id>>`
* `DELETE /{subject}/{format}/{version}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject-format-version>>`
* `DELETE /schemas/{id}` -- see `<<spring-cloud-stream-overview-deleting-schema-id>>`
* `DELETE /{subject}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject>>`
[[spring-cloud-stream-overview-registering-new-schema]]
===== Registering a New Schema
To register a new schema, send a `POST` request to the `/` endpoint.
The `/` accepts a JSON payload with the following fields:
* `subject`: The schema subject
* `format`: The schema format
* `definition`: The schema definition
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-subject-format-version]]
===== Retrieving an Existing Schema by Subject, Format, and Version
To retrieve an existing schema by subject, format, and version, send `GET` request to the `/{subject}/{format}/{version}` endpoint.
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-subject-format]]
===== Retrieving an Existing Schema by Subject and Format
To retrieve an existing schema by subject and format, send a `GET` request to the `/subject/format` endpoint.
Its response is a list of schemas with each schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-retrieve-schema-id]]
===== Retrieving an Existing Schema by ID
To retrieve a schema by its ID, send a `GET` request to the `/schemas/{id}` endpoint.
Its response is a schema object in JSON, with the following fields:
* `id`: The schema ID
* `subject`: The schema subject
* `format`: The schema format
* `version`: The schema version
* `definition`: The schema definition
[[spring-cloud-stream-overview-deleting-schema-subject-format-version]]
===== Deleting a Schema by Subject, Format, and Version
To delete a schema identified by its subject, format, and version, send a `DELETE` request to the `/{subject}/{format}/{version}` endpoint.
[[spring-cloud-stream-overview-deleting-schema-id]]
===== Deleting a Schema by ID
To delete a schema by its ID, send a `DELETE` request to the `/schemas/{id}` endpoint.
[[spring-cloud-stream-overview-deleting-schema-subject]]
===== Deleting a Schema by Subject
`DELETE /{subject}`
Delete existing schemas by their subject.
NOTE: This note applies to users of Spring Cloud Stream 1.1.0.RELEASE only.
Spring Cloud Stream 1.1.0.RELEASE used the table name, `schema`, for storing `Schema` objects. `Schema` is a keyword in a number of database implementations.
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.
If you want to use the Confluent schema registry, you need to create a bean of type `ConfluentSchemaRegistryClient`, which supersedes the one configured by default by the framework. The following example shows how to create such a bean:
[source,java]
----
@Bean
public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){
ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
client.setEndpoint(endpoint);
return client;
}
----
NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0.
=== Schema Registration and Resolution
To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections:
* `<<spring-cloud-stream-overview-schema-registration-process>>`
* `<<spring-cloud-stream-overview-schema-resolution-process>>`
[[spring-cloud-stream-overview-schema-registration-process]]
==== 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.
Avro types such as `SpecificRecord` or `GenericRecord` already contain a schema, which can be retrieved immediately from the instance.
In the case of POJOs, a schema is inferred if the `spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled` property is set to `true` (the default).
.Schema Writer Resolution Process
image::{github-raw}/docs/src/main/asciidoc/images/schema_resolution.png[width=800,scaledwidth="75%",align="center"]
Ones a schema is obtained, the converter loads its metadata (version) from the remote server.
First, it queries a local cache. If no result is found, it submits the data to the server, which replies with versioning information.
The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.
.Schema Registration Process
image::{github-raw}/docs/src/main/asciidoc/images/registration.png[width=800,scaledwidth="75%",align="center"]
With the schema version information, the converter sets the `contentType` header of the message to carry the version information -- for example: `application/vnd.user.v1+avro`.
[[spring-cloud-stream-overview-schema-resolution-process]]
==== Schema Resolution Process (Deserialization)
When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `<<spring-cloud-stream-overview-schema-registration-process>>`, the converter queries the Schema server to fetch the writer schema of the message.
Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro's schema resolution support, reads it into the reader definition (setting defaults and any missing properties).
.Schema Reading Resolution Process
image::{github-raw}/docs/src/main/asciidoc/images/schema_reading.png[width=800,scaledwidth="75%",align="center"]
NOTE: You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application).
We suggest taking a moment to read https://avro.apache.org/docs/1.7.6/spec.html[the Avro terminology] and understand the process.
Spring Cloud Stream always fetches the writer schema to determine how to read a message.
If you want to get Avro's schema evolution support working, you need to make sure that a `readerSchema` was properly set for your application.

37
docs/src/main/ruby/generate_readme.sh Executable file
View File

@@ -0,0 +1,37 @@
#!/usr/bin/env ruby
base_dir = File.join(File.dirname(__FILE__),'../../..')
src_dir = File.join(base_dir, "/src/main/asciidoc")
require 'asciidoctor'
require 'optparse'
options = {}
file = "#{src_dir}/README.adoc"
OptionParser.new do |o|
o.on('-o OUTPUT_FILE', 'Output file (default is stdout)') { |file| options[:to_file] = file unless file=='-' }
o.on('-h', '--help') { puts o; exit }
o.parse!
end
file = ARGV[0] if ARGV.length>0
# Copied from https://github.com/asciidoctor/asciidoctor-extensions-lab/blob/master/scripts/asciidoc-coalescer.rb
doc = Asciidoctor.load_file file, safe: :unsafe, header_only: true, attributes: options[:attributes]
header_attr_names = (doc.instance_variable_get :@attributes_modified).to_a
header_attr_names.each {|k| doc.attributes[%(#{k}!)] = '' unless doc.attr? k }
attrs = doc.attributes
attrs['allow-uri-read'] = true
puts attrs
out = "// Do not edit this file (e.g. go instead to src/main/asciidoc)\n\n"
doc = Asciidoctor.load_file file, safe: :unsafe, parse: false, attributes: attrs
out << doc.reader.read
unless options[:to_file]
puts out
else
File.open(options[:to_file],'w+') do |file|
file.write(out)
end
end