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

Support for custom avro schema de/serializer

Browse Source 
incorporating comments

incorporating unit test

incorporating unit test
This commit is contained in:
Ish Mahajan
2019-02-09 22:15:29 +05:30
committed by Oleg Zhurakousky
parent 73a8ffdefb
commit cb1c20ffa1
12 changed files with 570 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-stream-schema/pom.xml
View File

@@ -39,6 +39,10 @@
<version>${avro.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
@@ -54,6 +58,11 @@
<artifactId>spring-cloud-stream-schema-server</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-avro</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>

110
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java
View File

@@ -22,20 +22,9 @@ import java.util.Collection;
import java.util.Collections;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecord;
import org.springframework.core.io.Resource;
import org.springframework.messaging.Message;
@@ -58,14 +47,31 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
* common parser will let user to import external schemas.
*/
private Schema.Parser schemaParser = new Schema.Parser();
private AvroSchemaServiceManager avroSchemaServiceManager;
@Deprecated
protected AbstractAvroMessageConverter(MimeType supportedMimeType) {
this(Collections.singletonList(supportedMimeType));
this(Collections.singletonList(supportedMimeType), new AvroSchemaServiceManagerImpl());
}
protected AbstractAvroMessageConverter(MimeType supportedMimeType, AvroSchemaServiceManager avroSchemaServiceManager) {
this(Collections.singletonList(supportedMimeType), avroSchemaServiceManager);
}
@Deprecated
protected AbstractAvroMessageConverter(Collection<MimeType> supportedMimeTypes) {
this(supportedMimeTypes, new AvroSchemaServiceManagerImpl());
setContentTypeResolver(new OriginalContentTypeResolver());
}
protected AbstractAvroMessageConverter(Collection<MimeType> supportedMimeTypes, AvroSchemaServiceManager manager) {
super(supportedMimeTypes);
setContentTypeResolver(new OriginalContentTypeResolver());
this.avroSchemaServiceManager = manager;
}
protected AvroSchemaServiceManager avroSchemaServiceManager() {
return this.avroSchemaServiceManager;
}
protected Schema parseSchema(Resource r) throws IOException {
@@ -81,7 +87,7 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass,
Object conversionHint) {
Object result = null;
Object result;
try {
byte[] payload = (byte[]) message.getPayload();
@@ -98,11 +104,7 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
Schema writerSchema = resolveWriterSchemaForDeserialization(mimeType);
Schema readerSchema = resolveReaderSchemaForDeserialization(targetClass);
@SuppressWarnings("unchecked")
DatumReader<Object> reader = getDatumReader((Class<Object>) targetClass,
readerSchema, writerSchema);
Decoder decoder = DecoderFactory.get().binaryDecoder(payload, null);
result = reader.read(null, decoder);
result = avroSchemaServiceManager().readData(targetClass, payload, readerSchema, writerSchema);
}
catch (IOException e) {
throw new MessageConversionException(message, "Failed to read payload", e);
@@ -110,74 +112,6 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
return result;
}
private DatumWriter<Object> getDatumWriter(Class<Object> type, Schema schema) {
DatumWriter<Object> writer;
this.logger.debug("Finding correct DatumWriter for type " + type.getName());
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
writer = new SpecificDatumWriter<>(schema);
}
else {
writer = new SpecificDatumWriter<>(type);
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
writer = new GenericDatumWriter<>(schema);
}
else {
if (schema != null) {
writer = new ReflectDatumWriter<>(schema);
}
else {
writer = new ReflectDatumWriter<>(type);
}
}
return writer;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
protected DatumReader<Object> getDatumReader(Class<Object> type, Schema schema,
Schema writerSchema) {
DatumReader<Object> reader = null;
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new SpecificDatumReader<>(writerSchema, schema);
}
else {
reader = new SpecificDatumReader<>(schema);
}
}
else {
reader = new SpecificDatumReader<>(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new GenericDatumReader<>(writerSchema, schema);
}
else {
reader = new GenericDatumReader<>(schema);
}
}
}
else {
reader = new ReflectDatumReader(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
if (reader == null) {
throw new MessageConversionException("No schema can be inferred from type "
+ type.getName() + " and no schema has been explicitly configured.");
}
return reader;
}
@Override
protected Object convertToInternal(Object payload, MessageHeaders headers,
Object conversionHint) {
@@ -189,8 +123,8 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
}
Schema schema = resolveSchemaForWriting(payload, headers, hintedContentType);
@SuppressWarnings("unchecked")
DatumWriter<Object> writer = getDatumWriter(
(Class<Object>) payload.getClass(), schema);
DatumWriter<Object> writer = avroSchemaServiceManager()
.getDatumWriter(payload.getClass(), schema);
Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null);
writer.write(payload, encoder);
encoder.flush();

6
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java
View File

@@ -30,6 +30,7 @@ import org.springframework.cloud.stream.annotation.StreamMessageConverter;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
@@ -43,10 +44,13 @@ import org.springframework.util.ReflectionUtils;
@ConditionalOnProperty(value = "spring.cloud.stream.schemaRegistryClient.enabled", matchIfMissing = true)
@ConditionalOnBean(type = "org.springframework.cloud.stream.schema.client.SchemaRegistryClient")
@EnableConfigurationProperties({ AvroMessageConverterProperties.class })
@Import(AvroSchemaServiceManagerImpl.class)
public class AvroMessageConverterAutoConfiguration {
@Autowired
private AvroMessageConverterProperties avroMessageConverterProperties;
@Autowired
private AvroSchemaServiceManager avroSchemaServiceManager;
@Bean
@ConditionalOnMissingBean(AvroSchemaRegistryClientMessageConverter.class)
@@ -55,7 +59,7 @@ public class AvroMessageConverterAutoConfiguration {
SchemaRegistryClient schemaRegistryClient) {
AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter;
avroSchemaRegistryClientMessageConverter = new AvroSchemaRegistryClientMessageConverter(
schemaRegistryClient, cacheManager());
schemaRegistryClient, cacheManager(), avroSchemaServiceManager);
avroSchemaRegistryClientMessageConverter.setDynamicSchemaGenerationEnabled(
this.avroMessageConverterProperties.isDynamicSchemaGenerationEnabled());
if (this.avroMessageConverterProperties.getReaderSchema() != null) {

33
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java
View File

@@ -44,29 +44,62 @@ public class AvroSchemaMessageConverter extends AbstractAvroMessageConverter {
* Create a {@link AvroSchemaMessageConverter}. Uses the default {@link MimeType} of
* {@code "application/avro"}.
*/
@Deprecated
public AvroSchemaMessageConverter() {
super(new MimeType("application", "avro"));
}
/**
* Create a {@link AvroSchemaMessageConverter}. Uses the default {@link MimeType} of
* {@code "application/avro"}.
* @param manager for schema management
*/
public AvroSchemaMessageConverter(AvroSchemaServiceManager manager) {
super(new MimeType("application", "avro"), manager);
}
/**
* Create a {@link AvroSchemaMessageConverter}. The converter will be used for the
* provided {@link MimeType}.
* @param supportedMimeType mime type to be supported by
* {@link AvroSchemaMessageConverter}
*/
@Deprecated
public AvroSchemaMessageConverter(MimeType supportedMimeType) {
super(supportedMimeType);
}
/**
* Create a {@link AvroSchemaMessageConverter}. The converter will be used for the
* provided {@link MimeType}.
* @param supportedMimeType mime type to be supported by
* {@link AvroSchemaMessageConverter}
* @param manager for schema management
*/
public AvroSchemaMessageConverter(MimeType supportedMimeType, AvroSchemaServiceManager manager) {
super(supportedMimeType, manager);
}
/**
* Create a {@link AvroSchemaMessageConverter}. The converter will be used for the
* provided {@link MimeType}s.
* @param supportedMimeTypes the mime types supported by this converter
*/
@Deprecated
public AvroSchemaMessageConverter(Collection<MimeType> supportedMimeTypes) {
super(supportedMimeTypes);
}
/**
* Create a {@link AvroSchemaMessageConverter}. The converter will be used for the
* provided {@link MimeType}s.
* @param supportedMimeTypes the mime types supported by this converter
* @param manager for schema management
*/
public AvroSchemaMessageConverter(Collection<MimeType> supportedMimeTypes, AvroSchemaServiceManager manager) {
super(supportedMimeTypes, manager);
}
public Schema getSchema() {
return this.schema;
}

30
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java
View File

@@ -26,7 +26,6 @@ import java.util.stream.Stream;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericContainer;
import org.apache.avro.reflect.ReflectData;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanInitializationException;
@@ -115,6 +114,9 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
public static final MimeType DEFAULT_AVRO_MIME_TYPE = new MimeType("application",
"*+" + AVRO_FORMAT);
private static final AvroSchemaServiceManager defaultAvroSchemaServiceManager =
new AvroSchemaServiceManagerImpl();
private final CacheManager cacheManager;
protected Resource[] schemaImports = new Resource[] {};
@@ -141,15 +143,35 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
* @param cacheManager instance of {@link CacheManager} to cache parsed schemas. If
* caching is not required use {@link NoOpCacheManager}
*/
@Deprecated
public AvroSchemaRegistryClientMessageConverter(
SchemaRegistryClient schemaRegistryClient, CacheManager cacheManager) {
super(Collections.singletonList(DEFAULT_AVRO_MIME_TYPE));
SchemaRegistryClient schemaRegistryClient, CacheManager cacheManager) {
super(Collections.singletonList(DEFAULT_AVRO_MIME_TYPE), defaultAvroSchemaServiceManager);
Assert.notNull(schemaRegistryClient, "cannot be null");
Assert.notNull(cacheManager, "'cacheManager' cannot be null");
this.schemaRegistryClient = schemaRegistryClient;
this.cacheManager = cacheManager;
}
/**
* Creates a new instance, configuring it with {@link SchemaRegistryClient} and
* {@link CacheManager}.
* @param schemaRegistryClient the {@link SchemaRegistryClient} used to interact with
* the schema registry server.
* @param cacheManager instance of {@link CacheManager} to cache parsed schemas. If
* caching is not required use {@link NoOpCacheManager}
* @param manager instance of {@link AvroSchemaServiceManager} to manage schemas.
*/
public AvroSchemaRegistryClientMessageConverter(
SchemaRegistryClient schemaRegistryClient, CacheManager cacheManager, AvroSchemaServiceManager manager) {
super(Collections.singletonList(DEFAULT_AVRO_MIME_TYPE), manager);
Assert.notNull(schemaRegistryClient, "cannot be null");
Assert.notNull(cacheManager, "'cacheManager' cannot be null");
Assert.notNull(manager, "'avroSchemaServiceManager' cannot be null");
this.schemaRegistryClient = schemaRegistryClient;
this.cacheManager = cacheManager;
}
public boolean isDynamicSchemaGenerationEnabled() {
return this.dynamicSchemaGenerationEnabled;
}
@@ -354,7 +376,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
payload.getClass()));
}
else {
schema = ReflectData.get().getSchema(payload.getClass());
schema = super.avroSchemaServiceManager().getSchema(payload.getClass());
}
this.getCache(REFLECTION_CACHE_NAME)
.put(payload.getClass().getName(), schema);

75
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManager.java Normal file
View File

@@ -0,0 +1,75 @@
/*
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import java.io.IOException;
import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
/**
* Manage a {@link Schema} together with its String representation.
*
* Helps to substitute the default implementation of {@link org.apache.avro.Schema}
* Generation using Custom Avro schema generator
*
* Provide a custom bean definition of {@link AvroSchemaServiceManager} and mark
* it as @Primary to override the default implementation
*
* @author 5aab
*
*/
public interface AvroSchemaServiceManager {
/**
* get {@link Schema}.
* @param clazz {@link Class} for which schema generation is required
* @return returns avro schema for given class
*/
Schema getSchema(Class<?> clazz);
/**
* get {@link DatumWriter}.
* @param type {@link Class} of java object which needs to be serialized
* @param schema {@link Schema} of object which needs to be serialized
* @return datum writer which can be used to write Avro payload
*/
DatumWriter<Object> getDatumWriter(Class<? extends Object> type, Schema schema);
/**
* get {@link DatumReader}.
* @param type {@link Class} of java object which needs to be serialized
* @param schema {@link Schema} default schema of object which needs to be de-serialized
* @param writerSchema {@link Schema} writerSchema provided at run time
* @return datum reader which can be used to read Avro payload
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
DatumReader<Object> getDatumReader(Class<? extends Object> type, Schema schema, Schema writerSchema);
/**
* read data from avro type payload {@link DatumReader}.
* @param targetClass {@link Class} of java object which needs to be serialized
* @param payload {@link byte} serialized payload of object which needs to be de-serialized
* @param readerSchema {@link Schema} readerSchema of object which needs to be de-serialized
* @param writerSchema {@link Schema} writerSchema used to while serializing payload
* @return java object after reading Avro Payload
* @throws IOException in case of error
*/
Object readData(Class<? extends Object> targetClass, byte[] payload, Schema readerSchema, Schema writerSchema)
throws IOException;
}

169
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManagerImpl.java Normal file
View File

@@ -0,0 +1,169 @@
/*
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import java.io.IOException;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecord;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.stereotype.Component;
/**
* Default Concrete implementation of {@link AvroSchemaServiceManager}.
*
* Helps to substitute the default implementation of {@link org.apache.avro.Schema}
* Generation using Custom Avro schema generator
*
* Provide a custom bean definition of {@link AvroSchemaServiceManager} and mark
* it as @Primary to override this default implementation
*
* @author 5aab
*
*/
@Component
public class AvroSchemaServiceManagerImpl implements AvroSchemaServiceManager {
protected final Log logger = LogFactory.getLog(this.getClass());
/**
* get {@link Schema}.
* @param clazz {@link Class} for which schema generation
* is required
* @return returns avro schema for given class
*/
@Override
public Schema getSchema(Class<?> clazz) {
return ReflectData.get().getSchema(clazz);
}
/**
* get {@link DatumWriter}.
* @param type {@link Class} of java object which needs to be serialized
* @param schema {@link Schema} of object which needs to be serialized
* @return datum writer which can be used to write Avro payload
*/
@Override
public DatumWriter<Object> getDatumWriter(Class<?> type, Schema schema) {
DatumWriter<Object> writer;
this.logger.debug("Finding correct DatumWriter for type " + type.getName());
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
writer = new SpecificDatumWriter<>(schema);
}
else {
writer = new SpecificDatumWriter(type);
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
writer = new GenericDatumWriter<>(schema);
}
else {
if (schema != null) {
writer = new ReflectDatumWriter<>(schema);
}
else {
writer = new ReflectDatumWriter(type);
}
}
return writer;
}
/**
* get {@link DatumReader}.
* @param type {@link Class} of java object which needs to be serialized
* @param schema {@link Schema} default schema of object which needs to be de-serialized
* @param writerSchema {@link Schema} writerSchema provided at run time
* @return datum reader which can be used to read Avro payload
*/
@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public DatumReader<Object> getDatumReader(Class<?> type, Schema schema, Schema writerSchema) {
DatumReader<Object> reader = null;
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new SpecificDatumReader<>(writerSchema, schema);
}
else {
reader = new SpecificDatumReader<>(schema);
}
}
else {
reader = new SpecificDatumReader(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new GenericDatumReader<>(writerSchema, schema);
}
else {
reader = new GenericDatumReader<>(schema);
}
}
}
else {
reader = new ReflectDatumReader(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
if (reader == null) {
throw new MessageConversionException("No schema can be inferred from type "
+ type.getName() + " and no schema has been explicitly configured.");
}
return reader;
}
/**
* read data from avro type payload {@link DatumReader}.
* @param clazz {@link Class} of java object which needs to be serialized
* @param payload {@link byte} serialized payload of object which needs to be de-serialized
* @param readerSchema {@link Schema} readerSchema of object which needs to be de-serialized
* @param writerSchema {@link Schema} writerSchema used to while serializing payload
* @return java object after reading Avro Payload
* @throws IOException is thrown in case of error
*/
@Override
public Object readData(Class<? extends Object> clazz, byte[] payload, Schema readerSchema,
Schema writerSchema) throws IOException {
DatumReader<Object> reader = this.getDatumReader(clazz,
readerSchema, writerSchema);
Decoder decoder = DecoderFactory.get().binaryDecoder(payload, null);
return reader.read(null, decoder);
}
}

11
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java
View File

@@ -44,6 +44,8 @@ import org.springframework.cache.support.NoOpCacheManager;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManager;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManagerImpl;
import org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy;
import org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
@@ -123,8 +125,9 @@ public class AvroMessageConverterSerializationTests {
@Test
public void testSchemaImport() throws Exception {
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
client, new NoOpCacheManager(), manager);
converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy());
converter.setDynamicSchemaGenerationEnabled(false);
converter.setSchemaLocations(this.schemaRegistryServerContext
@@ -150,8 +153,9 @@ public class AvroMessageConverterSerializationTests {
GenericRecord genericRecord = new GenericData.Record(v1);
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
client, new NoOpCacheManager(), manager);
converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy());
converter.setDynamicSchemaGenerationEnabled(false);
@@ -183,8 +187,9 @@ public class AvroMessageConverterSerializationTests {
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
client.register("user", "avro", v1.toString());
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
client, new NoOpCacheManager(), manager);
converter.setDynamicSchemaGenerationEnabled(false);
converter.afterPropertiesSet();
ByteArrayOutputStream baos = new ByteArrayOutputStream();

8
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java
View File

@@ -33,6 +33,8 @@ import org.springframework.cloud.stream.annotation.StreamMessageConverter;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManager;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManagerImpl;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
@@ -222,8 +224,9 @@ public class AvroSchemaMessageConverterTests {
@Bean
@StreamMessageConverter
public MessageConverter userMessageConverter() throws IOException {
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
MimeType.valueOf("avro/bytes"));
MimeType.valueOf("avro/bytes"), manager);
if (this.schemaLocation != null) {
avroSchemaMessageConverter.setSchemaLocation(this.schemaLocation);
}
@@ -253,8 +256,9 @@ public class AvroSchemaMessageConverterTests {
@Bean
@StreamMessageConverter
public MessageConverter userMessageConverter() throws IOException {
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
MimeType.valueOf("avro/bytes"));
MimeType.valueOf("avro/bytes"), manager);
if (this.schemaLocation != null) {
avroSchemaMessageConverter.setSchemaLocation(this.schemaLocation);
}

5
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java
View File

@@ -38,6 +38,8 @@ import org.springframework.cloud.stream.annotation.StreamMessageConverter;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManager;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManagerImpl;
import org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient;
import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
@@ -239,8 +241,9 @@ public class AvroSchemaRegistryClientMessageConverterTests {
@Bean
@StreamMessageConverter
AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() {
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
return new AvroSchemaRegistryClientMessageConverter(
new DefaultSchemaRegistryClient(), new NoOpCacheManager());
new DefaultSchemaRegistryClient(), new NoOpCacheManager(), manager);
}
@Bean

173
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaServiceManagerTests.java Normal file
View File

@@ -0,0 +1,173 @@
/*
* Copyright 2017-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import java.io.File;
import java.io.IOException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.avro.AvroFactory;
import com.fasterxml.jackson.dataformat.avro.AvroMapper;
import com.fasterxml.jackson.dataformat.avro.AvroSchema;
import com.fasterxml.jackson.dataformat.avro.schema.AvroSchemaGenerator;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.avro.SchemaParseException;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.assertj.core.util.Lists;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.cloud.schema.avro.domain.FoodOrder;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManager;
import org.springframework.cloud.stream.schema.avro.AvroSchemaServiceManagerImpl;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.util.MimeType;
/**
* @author 5aab
*/
@Slf4j
public class AvroSchemaServiceManagerTests {
@Test(expected = DataFileWriter.AppendWriteException.class)
public void testWithDefaultImplementation() throws IOException {
AvroSchemaServiceManager defaultServiceManager = new AvroSchemaServiceManagerImpl();
Schema schema = defaultServiceManager.getSchema(FoodOrder.class);
FoodOrder foodOrder = FoodOrder.builder().restaurant("Spring Kitchen")
.orderDescription("avro makhani").customerAddress("world wide web").build();
File file = new File("foodorder.avro");
DatumWriter datumWriter = defaultServiceManager.getDatumWriter(foodOrder.getClass(), schema);
DataFileWriter<FoodOrder> dataFileWriter = new DataFileWriter<FoodOrder>(datumWriter);
dataFileWriter.create(schema, file);
dataFileWriter.append(foodOrder);
FoodOrder foodOrder2 = FoodOrder.builder().restaurant(null)
.orderDescription(null).customerAddress(null).build();
dataFileWriter.append(foodOrder2);
dataFileWriter.close();
DatumReader userDatumReader = defaultServiceManager.getDatumReader(foodOrder.getClass(), schema, schema);
DataFileReader<FoodOrder> dataFileReader = new DataFileReader<FoodOrder>(file, userDatumReader);
FoodOrder foodOrderDeserialized = null;
while (dataFileReader.hasNext()) {
// Reuse user object by passing it to next(). This saves us from
// allocating and garbage collecting many objects for files with
// many items.
foodOrderDeserialized = dataFileReader.next(foodOrderDeserialized);
System.out.println("De-serialised Successfully : " + foodOrderDeserialized);
}
}
@Test
public void testWithCustomImplementation() throws IOException {
AvroSchemaServiceManager manager = new AvroSchemaServiceManager() {
@Override
public Schema getSchema(Class<?> clazz) {
ObjectMapper mapper = new ObjectMapper(new AvroFactory());
AvroSchemaGenerator gen = new AvroSchemaGenerator();
try {
mapper.acceptJsonFormatVisitor(FoodOrder.class, gen);
}
catch (JsonMappingException e) {
log.error("Error while setting acceptJsonFormatVisitor {}", e);
}
AvroSchema schemaWrapper = gen.getGeneratedSchema();
return schemaWrapper.getAvroSchema();
}
@Override
public DatumWriter<Object> getDatumWriter(Class<?> type, Schema schema) {
return new AvroSchemaServiceManagerImpl().getDatumWriter(type, schema);
}
@Override
public DatumReader<Object> getDatumReader(Class<?> type, Schema schema, Schema writerSchema) {
return new AvroSchemaServiceManagerImpl().getDatumReader(type, schema, schema);
}
@Override
public Object readData(Class<? extends Object> targetClass, byte[] payload, Schema readerSchema,
Schema writerSchema) throws IOException {
ObjectMapper mapper = new ObjectMapper(new AvroFactory());
AvroSchemaGenerator gen = new AvroSchemaGenerator();
try {
mapper.acceptJsonFormatVisitor(targetClass, gen);
}
catch (JsonMappingException e) {
log.error("Error while setting acceptJsonFormatVisitor {}", e);
}
return mapper.readerFor(targetClass)
.with(new AvroSchema(readerSchema))
.readValue(payload);
}
};
FoodOrder foodOrder1 = FoodOrder.builder().restaurant("Spring Kitchen")
.orderDescription("avro makhani").customerAddress("world wide web").build();
FoodOrder foodOrder2 = FoodOrder.builder().restaurant("Spring Kitchen")
.orderDescription(null).customerAddress(null).build();
Schema schema = manager.getSchema(FoodOrder.class);
AvroMapper mapper = new AvroMapper();
byte[] payload1 = mapper.writer(new AvroSchema(schema)).writeValueAsBytes(foodOrder1);
byte[] payload2 = mapper.writer(new AvroSchema(schema)).writeValueAsBytes(foodOrder2);
foodOrder1 = (FoodOrder) manager.readData(foodOrder1.getClass(), payload1, schema, schema);
foodOrder2 = (FoodOrder) manager.readData(foodOrder1.getClass(), payload2, schema, schema);
Assert.assertNull(foodOrder2.getOrderDescription());
Assert.assertNull(foodOrder2.getCustomerAddress());
}
@Test
public void testAvroSchemaMessageConverter() {
AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter();
MimeType mimeType = new MimeType("application", "avro");
Assert.assertEquals(converter.getSupportedMimeTypes().get(0), mimeType);
AvroSchemaMessageConverter converter2 = new AvroSchemaMessageConverter(mimeType);
Assert.assertEquals(converter2.getSupportedMimeTypes().get(0), mimeType);
AvroSchemaMessageConverter converter3 =
new AvroSchemaMessageConverter(Lists.newArrayList(mimeType));
Assert.assertEquals(converter3.getSupportedMimeTypes().get(0), mimeType);
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaMessageConverter converter4 = new AvroSchemaMessageConverter(manager);
Assert.assertEquals(converter4.getSupportedMimeTypes().get(0), mimeType);
AvroSchemaMessageConverter converter5 =
new AvroSchemaMessageConverter(Lists.newArrayList(mimeType), manager);
Schema schema = manager.getSchema(FoodOrder.class);
converter5.setSchema(schema);
Assert.assertEquals(converter5.getSupportedMimeTypes().get(0), mimeType);
Assert.assertEquals(converter5.getSchema(), schema);
}
@Test(expected = SchemaParseException.class)
public void testAvroSchemaMessageConverterException() {
MimeType mimeType = new MimeType("application", "avro");
AvroSchemaServiceManager manager = new AvroSchemaServiceManagerImpl();
AvroSchemaMessageConverter converter =
new AvroSchemaMessageConverter(Lists.newArrayList(mimeType), manager);
converter.setSchemaLocation(new ByteArrayResource(new byte[2]) {
});
}
}

40
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/domain/FoodOrder.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2017-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;
/**
* @author 5aab
*/
@Setter
@Getter
@AllArgsConstructor
@NoArgsConstructor
@Builder
@ToString
public class FoodOrder {
private String restaurant;
private String customerAddress;
private String orderDescription;
}