diff --git a/spring-cloud-stream-schema/pom.xml b/spring-cloud-stream-schema/pom.xml index 84f04ba5e..fee7e8100 100644 --- a/spring-cloud-stream-schema/pom.xml +++ b/spring-cloud-stream-schema/pom.xml @@ -39,6 +39,10 @@ ${avro.version} true + + org.projectlombok + lombok + org.springframework.cloud spring-cloud-stream-test-support @@ -54,6 +58,11 @@ spring-cloud-stream-schema-server test + + com.fasterxml.jackson.dataformat + jackson-dataformat-avro + test + diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java index 2ef6220c8..65cb102fe 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java @@ -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 supportedMimeTypes) { + this(supportedMimeTypes, new AvroSchemaServiceManagerImpl()); + setContentTypeResolver(new OriginalContentTypeResolver()); + } + + protected AbstractAvroMessageConverter(Collection 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 reader = getDatumReader((Class) 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 getDatumWriter(Class type, Schema schema) { - DatumWriter 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 getDatumReader(Class type, Schema schema, - Schema writerSchema) { - DatumReader 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 writer = getDatumWriter( - (Class) payload.getClass(), schema); + DatumWriter writer = avroSchemaServiceManager() + .getDatumWriter(payload.getClass(), schema); Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null); writer.write(payload, encoder); encoder.flush(); diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java index 4e4ba8441..1d828a9ee 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java @@ -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) { diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java index 8f073020c..5fc8863e9 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java @@ -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 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 supportedMimeTypes, AvroSchemaServiceManager manager) { + super(supportedMimeTypes, manager); + } + public Schema getSchema() { return this.schema; } diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java index 4bece7a05..4330fb3a5 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java @@ -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); diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManager.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManager.java new file mode 100644 index 000000000..400f28ae4 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManager.java @@ -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 getDatumWriter(Class 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 getDatumReader(Class 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 targetClass, byte[] payload, Schema readerSchema, Schema writerSchema) + throws IOException; +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManagerImpl.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManagerImpl.java new file mode 100644 index 000000000..888f40eb1 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaServiceManagerImpl.java @@ -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 getDatumWriter(Class type, Schema schema) { + DatumWriter 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 getDatumReader(Class type, Schema schema, Schema writerSchema) { + DatumReader 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 clazz, byte[] payload, Schema readerSchema, + Schema writerSchema) throws IOException { + DatumReader reader = this.getDatumReader(clazz, + readerSchema, writerSchema); + Decoder decoder = DecoderFactory.get().binaryDecoder(payload, null); + return reader.read(null, decoder); + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java index 6eb976b55..8dbdb271e 100644 --- a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java @@ -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(); diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java index 5d4675b87..c63d629a6 100644 --- a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java @@ -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); } diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java index 85cab986f..4893eacdd 100644 --- a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java @@ -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 diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaServiceManagerTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaServiceManagerTests.java new file mode 100644 index 000000000..bad713705 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaServiceManagerTests.java @@ -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 dataFileWriter = new DataFileWriter(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 dataFileReader = new DataFileReader(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 getDatumWriter(Class type, Schema schema) { + return new AvroSchemaServiceManagerImpl().getDatumWriter(type, schema); + } + + @Override + public DatumReader getDatumReader(Class type, Schema schema, Schema writerSchema) { + return new AvroSchemaServiceManagerImpl().getDatumReader(type, schema, schema); + } + + @Override + public Object readData(Class 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]) { + }); + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/domain/FoodOrder.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/domain/FoodOrder.java new file mode 100644 index 000000000..5d032bb94 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/domain/FoodOrder.java @@ -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; +}